Storm整合Kafka

package com.zj.storm.kafka; import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; import org.apache.storm.kafka.spout.KafkaSpout; import org.apache.storm.kafka.spout.KafkaSpoutConfig; import org.apache.storm.topology.TopologyBuilder; /** * App */ public class App { public static void main(String[] args) throws Exception { TopologyBuilder topologyBuilder = new TopologyBuilder(); // 设置Spout // 通过kafka spout Config获得kafka spout // 设置5个线程接收数据 topologyBuilder.setSpout("kafka_spout", new KafkaSpout<>(KafkaSpoutConfig .builder("192.168.217.151:9092", "kafkatopic").setGroupId("default-group").build()), 2); topologyBuilder.setBolt("creator-bolt", new CallLogCreatorBolt(), 2).setNumTasks(2) .shuffleGrouping("kafka_spout"); Config config = new Config(); config.setDebug(true); config.setNumWorkers(2); //本地集群-测试 LocalCluster cluster = new LocalCluster(); cluster.submitTopology("local-storm", config, topologyBuilder.createTopology()); // Thread.sleep(10000); //停止集群 // cluster.shutdown(); //线上集群-生产 // StormSubmitter.submitTopology("topo-kafka", config, topologyBuilder.createTopology()); } }