如何实现Storm-kafka

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

         实现细节如下：

public static Iterable<List<Object>> generateTuples(	KafkaConfig kafkaConfig, Message msg) {	Iterable<List<Object>> 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<List<Object>> 类型的tups对象

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

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

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

     细节上，让我们查看一下  KeyValueSchemeAsMultiScheme 对象

package com.mixbox.storm.kafka; import backtype.storm.spout.SchemeAsMultiScheme; import java.util.Arrays; import java.util.List; @SuppressWarnings("serial") public class KeyValueSchemeAsMultiScheme extends SchemeAsMultiScheme{     public KeyValueSchemeAsMultiScheme(KeyValueScheme scheme) {         super(scheme);     }          //把一个key value 进行反序列化     public Iterable<List<Object>> deserializeKeyAndValue(final byte[] key, final byte[] value) {              List<Object> o = ((KeyValueScheme)scheme).deserializeKeyAndValue(key, value);                  if(o == null) return null;                  else return Arrays.asList(o);     } }

           首先， KeyValueSchemeAsMultiScheme 继承自SchemeAsMultiScheme，SchemeAsMultiScheme是Storm自身所带有的一个class实现，以上的2个class 都实现了Storm之中【MultiScheme】

关于“如何实现Storm-kafka”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，使各位可以学到更多知识，如果觉得文章不错，请把它分享出去让更多的人看到。