public class KafkaTopo {

	public static void main(String[] args) {

		String zkRoot = "/kafka-storm";

		String spoutId = "KafkaSpout";

		BrokerHosts brokerHosts = new ZkHosts("m2:2181,m7:2181,m8:2181");

		SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "test2", zkRoot, spoutId);

		// spoutConfig.forceFromStart = true;

		spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme());

		TopologyBuilder builder = new TopologyBuilder();

		//设置一个spout用来从kaflka消息队列中读取数据并发送给下一级的bolt组件，此处用的spout组件并非自定义的，而是storm中已经开发好的KafkaSpout

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

		builder.setBolt("UpperBolt", new UpperBolt()).shuffleGrouping("KafkaSpout");

		builder.setBolt("ExtBolt", new ExtBolt(), 4).fieldsGrouping("UpperBolt", new Fields("name"));

		Config conf = new Config();

		conf.setNumWorkers(4);

		conf.setNumAckers(0);

		conf.setDebug(false);

		//LocalCluster用来将topology提交到本地模拟器运行，方便开发调试

		LocalCluster cluster = new LocalCluster();

		cluster.submitTopology("WordCount", conf, builder.createTopology());

		//提交topology到storm集群中运行

//		StormSubmitter.submitTopology("sufei-topo", conf, builder.createTopology());

	}

}