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);

    }

}