import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import storm.kafka.*;public class TopologyWordCount {public static void main(String[] args) throws Exception {TopologyBuilder builder = new TopologyBuilder();// BrokerHosts接口有2个实现类StaticHosts和ZkHosts,ZkHosts会定时(默认60秒)从ZK中更新brokers的信息,StaticHosts是则不会// 要注意这里的第二个参数brokerZkPath要和kafka中的server.properties中配置的zookeeper.connect对应,没有专门配置就可以不填写,使用默认值BrokerHosts brokerHosts = new ZkHosts("192.168.0.171:2181," + "192.168.0.207:2181," + "192.168.0.204:2181");// 定义spoutConfig// 第一个参数hosts是上面定义的brokerHosts// 第二个参数topic是该Spout订阅的topic名称,需要提前创建这个topic// 第三个参数zkRoot是存储消费的offset(存储在zookeeper中了),也就是当这个Topology遇到故障重启后会将故障期间未消费的message继续消费而不会丢失(可不配置)// 第四个参数id是当前spout的唯一标识,当存在多个spout时,id不能冲突SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "shuaige33333", "", "0");// 定义kafkaSpout如何解析数据,这里是将kafka的producer send的数据放入到String// 类型的str变量中输出,这个str是StringSchema定义的变量名称/*通过字符串的方式解析数据*/spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());/*开始拓扑套路*/builder.setSpout("kafkaSpout", new KafkaSpout(spoutConfig));/*kafka 的spout*/builder.setBolt("WordSplitBolt", new WordSplitBolt()).shuffleGrouping("kafkaSpout");/*WordCountBolt获得从WordSplitBolt中传递过来的单词并统计词频*/builder.setBolt("WordCountBolt", new WordCountBolt()).fieldsGrouping("WordSplitBolt", new Fields("word"));// 本地运行或者提交到集群if (args != null && args.length == 1) {// 集群运行StormSubmitter.submitTopology(args[0], new Config(), builder.createTopology());} else {// 本地运行LocalCluster cluster = new LocalCluster();cluster.submitTopology("local", new Config(), builder.createTopology());Thread.sleep(10000000);cluster.shutdown();}}
}

只需要更改TopologyWordCount,就可以将storm集成kafka,将storm作为kafka的消费者,之后的那些逻辑就可以自己来写了.

builder.setSpout("kafkaSpout", new KafkaSpout(spoutConfig));

这里的KafkaSpout 是官方的Spout,在下一个bolt进行数据的提取时,要使用的方法是:

public void execute(Tuple input, BasicOutputCollector basicOutputCollector) {String line = input.getStringByField("str");//得到进入kafka的字符串for (String word : line.split(" ")) {basicOutputCollector.emit(new Values(word));}}

然后就可以测试了: