聊聊storm的direct grouping

序本文主要研究一下storm的direct groupingdirect groupingdirect grouping是一种特殊的grouping，它是由上游的producer直接指定下游哪个task去接收它发射出来的tuple。direct grouping的使用有如下几个步骤：1、上游在prepare方法保存下游bolt的taskId列表public class SentenceDirectBolt extends BaseRichBolt { private static final Logger LOGGER = LoggerFactory.getLogger(SentenceDirectBolt.class); private OutputCollector collector; private List<Integer> taskIds; private int numCounterTasks; public void prepare(Map config, TopologyContext context, OutputCollector collector) { this.collector = collector; //NOTE 1 这里要取到下游的bolt的taskId，用于emitDirect时指定taskId this.taskIds = context.getComponentTasks(“count-bolt”); this.numCounterTasks = taskIds.size(); } //……}这里保存了下游的bolt的taskId列表，用于emitDirect时选择taskId2、上游在declareOutputFields使用declareStream声明streamIdpublic class SentenceDirectBolt extends BaseRichBolt { //…… public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields(“word”)); //NOTE 2 这里要通过declareStream声明direct stream，并指定streamId declarer.declareStream(“directStreamDemo1”,true,new Fields(“word”)); declarer.declareStream(“directStreamDemo2”,true,new Fields(“word”)); }}这里声明了两个streamId，一个是directStreamDemo1，一个是directStreamDemo23、上游采用emitDirect指定下游taskId及streamIdpublic class SentenceDirectBolt extends BaseRichBolt { //…… public void execute(Tuple tuple) { String sentence = tuple.getStringByField(“sentence”); String[] words = sentence.split(" “); for(String word : words){ int targetTaskId = getWordCountTaskId(word); LOGGER.info(“word:{} choose taskId:{}",word,targetTaskId); // NOTE 3 这里指定发送给下游bolt的哪个taskId，同时指定streamId if(targetTaskId % 2 == 0){ this.collector.emitDirect(targetTaskId,“directStreamDemo1”,new Values(word)); }else{ this.collector.emitDirect(targetTaskId,“directStreamDemo2”,new Values(word)); } } this.collector.ack(tuple); }}这里使用emitDirect(int taskId, String streamId, List<Object> tuple)方法指定了下游的taskId以及要发送到的streamId4、下游使用directGrouping连接上游bolt及streamId @Test public void testDirectGrouping() throws InvalidTopologyException, AuthorizationException, AlreadyAliveException { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(“sentence-spout”, new SentenceSpout()); // SentenceSpout –> SplitSentenceBolt builder.setBolt(“split-bolt”, new SentenceDirectBolt()).shuffleGrouping(“sentence-spout”); // SplitSentenceBolt –> WordCountBolt //NOTE 4这里要指定上游的bolt以及要处理的streamId builder.setBolt(“count-bolt”, new WordCountBolt(),5).directGrouping(“split-bolt”,“directStreamDemo1”); // WordCountBolt –> ReportBolt builder.setBolt(“report-bolt”, new ReportBolt()).globalGrouping(“count-bolt”); submitRemote(builder); }这里count-bolt作为split-bolt的下游，使用了directGrouping，同时指定了要接收的streamId为directStreamDemo1小结direct grouping是一种特殊的grouping，它是由上游的producer直接指定下游哪个task去接收它发射出来的tuple。下游使用directGrouping连接上游同时指定要消费的streamId，上游在prepare的时候保存下游的taskId列表，然后在declareOutputFields的时候使用declareStream来声明streamId，最后在execute方法里头使用emitDirect(int taskId, String streamId, List<Object> tuple)方法指定了下游的taskId以及要发送到的streamIddocConceptsCommon Topology Patterns关于Storm Stream grouping