Storm的BaseBasicBolt源码解析ack机制

public class SplitSentenceBolt extends BaseBasicBolt { public void prepare(Map stormConf, TopologyContext context) { super.prepare(stormConf, context); } 
 //5：执行我们自己的逻辑处理方法，接收传入的参数。
 public void execute(Tuple input, BasicOutputCollector collector) { String sentence = (String)input.getValueByField("sentence"); String[] words = sentence.split(" "); for (String word : words) { word = word.trim(); word = word.toLowerCase(); collector.emit(new Values(word,1));//这个地方就是调用OutputCollector的包装类，来发消息 } } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word","num")); } }

public class BasicBoltExecutor implements IRichBolt { public static Logger LOG = LoggerFactory.getLogger(BasicBoltExecutor.class); private IBasicBolt _bolt; private transient BasicOutputCollector _collector;  //1：创建该对象，然后把我们写的SplitSentenceBolt对象赋给父类IBasicBolt。 public BasicBoltExecutor(IBasicBolt bolt) { _bolt = bolt; } public void declareOutputFields(OutputFieldsDeclarer declarer) { _bolt.declareOutputFields(declarer);//这里就是调用SplitSentenceBolt对象的方法了。 } //2:给BasicOutputCollector _collector字段赋值，BasicOutputCollector就是对OutputCollector类的包装。 public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { _bolt.prepare(stormConf, context); _collector = new BasicOutputCollector(collector); } //3：然后程序执行该方法，input的值source: spout1:4, stream: default, id: {}, [+ - * % /] public void execute(Tuple input) { _collector.setContext(input);//把接收到的tuple值设置给BasicOutputCollector中inputTuple字段。 try { _bolt.execute(input, _collector);//这个地方是调用我们实现类SplitSentenceBolt的ececute方法。 _collector.getOutputter().ack(input);//这个地方就是响应 } catch(FailedException e) { if(e instanceof ReportedFailedException) { _collector.reportError(e); } _collector.getOutputter().fail(input);//这个地方就是响应  } } public void cleanup() { _bolt.cleanup(); } public Map<String, Object> getComponentConfiguration() { return _bolt.getComponentConfiguration(); } }

public class BasicOutputCollector implements IBasicOutputCollector { private OutputCollector out; private Tuple inputTuple; public BasicOutputCollector(OutputCollector out) { this.out = out; }
 //4:把收到的tuple数据赋值给inputTuple，这个时候BasicOutputCollector对象的字段都具有值了。

 public void setContext(Tuple inputTuple) { this.inputTuple = inputTuple; }
 //6:这里我们发送新的（转换后的）tuple数据，看他内部的调用，其实他也会发送一个anchor tuple来保持tracker链路，
而这个anchor tuple就是bolt接收到转换前的源tuple数据。
 public List<Integer> emit(List<Object> tuple) { 
 return emit(Utils.DEFAULT_STREAM_ID, tuple); 
 } public List<Integer> emit(String streamId, List<Object> tuple) { return out.emit(streamId, inputTuple, tuple); } public void emitDirect(int taskId, String streamId, List<Object> tuple) { out.emitDirect(taskId, streamId, inputTuple, tuple); } public void emitDirect(int taskId, List<Object> tuple) { emitDirect(taskId, Utils.DEFAULT_STREAM_ID, tuple); } protected IOutputCollector getOutputter() { return out; } public void reportError(Throwable t) { out.reportError(t); } }