转自：http://blog.csdn.net/xeseo/article/details/18615761 有删改
A. 使用KafkaSpout
一个KafkaSpout只能去处理一个topic的内容，所以，它要求初始化时提供如下与topic相关信息：
- Kafka集群中的Broker地址 （IP+Port）
有两种方法指定：
1. 使用静态地址，即直接给定Kafka集群中所有Broker信息
-
GlobalPartitionInformationinfo=newGlobalPartitionInformation();
-
info.addPartition(0,newBroker("10.1.110.24",9092));
-
info.addPartition(0,newBroker("10.1.110.21",9092));
-
BrokerHostsbrokerHosts=newStaticHosts(info);
2. 从Zookeeper动态读取
-
BrokerHostsbrokerHosts=newZkHosts("10.1.110.24:2181,10.1.110.22:2181");
推荐使用这种方法，因为Kafka的Broker可能会动态的增减
- topic名字
- 当前spout的唯一标识Id （以下代称$spout_id）(应该是topic-ID)
- zookeeper上用于存储当前处理到哪个Offset了 （以下代称$zk_root(Zookeeper中存储offset的ZNode))
- 当前topic中数据如何解码
了解Kafka的应该知道，Kafka中当前处理到哪的Offset是由客户端自己管理的。所以，后面两个的目的，其实是在zookeeper上建立一个 $zk_root/$spout_id 的节点，其值是一个map，存放了当前Spout处理的Offset的信息。
在Topology中加入Spout的代码：
-
Stringtopic="test";
-
StringzkRoot="kafkastorm";
-
StringspoutId="myKafka";
-
-
SpoutConfigspoutConfig=newSpoutConfig(brokerHosts,topic,zkRoot,spoutId);
-
spoutConfig.scheme=newSchemeAsMultiScheme(newTestMessageScheme());
-
-
TopologyBuilderbuilder=newTopologyBuilder();
-
builder.setSpout("spout",newKafkaSpout(spoutConfig),spoutNum);
其中TestMessageScheme就是告诉KafkaSpout如何去解码数据，生成Storm内部传递数据
-
publicclassTestMessageSchemeimplementsScheme{
-
-
privatestaticfinalLoggerLOGGER=LoggerFactory.getLogger(TestMessageScheme.class);
-
-
@Override
-
publicList<Object>deserialize(byte[]bytes){
-
try{
-
Stringmsg=newString(bytes,"UTF-8");
-
returnnewValues(msg);
-
}catch(InvalidProtocolBufferExceptione){
-
LOGGER.error("Cannotparsetheprovidedmessage!");
-
}
-
-
-
returnnull;
-
}
-
-
@Override
-
publicFieldsgetOutputFields(){
-
returnnewFields("msg");
-
}
-
-
}
这个解码方式是与Producer端生成时，写入数据的编码方式配套的。这里我Producer端写入的是String的byte，所以这里也还原成String，定义输出为一个名叫"msg"的field。
后面就可以自己添加Bolt处理tuple中该field的数据了。
B.使用TransactionalTridentKafkaSpout
TransactionalTridentKafkaSpout是为事务性的Trident而用的。用法与KafkaSpout有所不同。
-
TridentKafkaConfigkafkaConfig=newTridentKafkaConfig(brokerHosts,topic,spoutId);
-
kafkaConfig.scheme=newSchemeAsMultiScheme(newTestMessageScheme());
-
-
TransactionalTridentKafkaSpoutkafkaSpout=newTransactionalTridentKafkaSpout(kafkaConfig);
-
-
TridentTopologytopology=newTridentTopology();
-
topology.newStream("test_str",kafkaSpout).shuffle().each(newFields("msg",newPrintFunction());
看到它并没有要求我们提供zkRoot，因为直接代码里面写死了…… -_-T
地址是 /transactional/<STREAM_NAME>/<Spout_Id>,在上面的例子中，就是 /transactional/test_str/myKafaka
C.常见问题
本地模式无法保存Offset
KafkaSpout初始化时，会去取spoutConfig.zkServers 和spoutConfig.zkPort 变量的值，而该值默认是空，那么它就会去取当前运行的Storm所配置的zookeeper地址和端口。而本地运行的Storm，是一个临时的zookeeper实例，并不会真正持久化。所以，每次关闭后，数据就没了。
本地模式，可以显示指定一个持续运行的Zookeeper用于存储当前spout在kafka中消费数据的offset。
-
spoutConfig.zkServers=newArrayList<String>(){{
-
add("10.1.110.20");
-
add("10.1.110.21");
-
add("10.1.110.24");
-
}};
-
spoutConfig.zkPort=2181;
