如何实现Storm-kafka

这篇文章将为大家详细讲解有关如何实现Storm-kafka，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。

实现细节如下：

public static Iterable> generateTuples(                        KafkaConfig kafkaConfig, Message msg) {                Iterable> tups;                ByteBuffer payload = msg.payload();                ByteBuffer key = msg.key();                if (key != null                                && kafkaConfig.scheme instanceof KeyValueSchemeAsMultiScheme) {                        tups = ((KeyValueSchemeAsMultiScheme) kafkaConfig.scheme)                                        .deserializeKeyAndValue(Utils.toByteArray(key),                                                        Utils.toByteArray(payload));                } else {                        tups = kafkaConfig.scheme.deserialize(Utils.toByteArray(payload));                }                return tups;        }

1 ：方法内部 Iterable> 类型的tups对象

2 ： Message对象是kafka的Message。 包括了Key，以及palyload负载

3 : 在生成元组的时候，需要拿到kafkaConfig.scheme 当中的Scheme，一旦Scheme instanceof

KeyValueSchemeAsMultiScheme，将强制性转换为 KeyValueSchemeAsMultiScheme，并且调用deserializeKeyAndValue 方法将其反序列化 List