public class SplitSentence implements IRichBolt {
        OutputCollector _collector;
 
        public void prepare(Map conf,
                         TopologyContext context,
                         OutputCollector collector) {
              _collector = collector;
        }
 
        public void execute(Tuple tuple) {
                String sentence = tuple.getString(0);
                for(String word: sentence.split(" ")) {
                      _collector.emit(tuple,new Values(word));
                }
             _collector.ack(tuple);
        }
 
       publicvoid cleanup() {}
       publicvoid declareOutputFields(OutputFieldsDeclarer declarer) {
                  declarer.declare(newFields("word"));
       }
   }

       你可以调用OutputCollector 的fail方法去立即将从消息源头发出的那个tuple标记为fail， 比如你查询了数据库，发现一个错误，你可以马上fail那个输入tuple， 这样可以让这个tuple被快速的重新处理， 因为你不需要等那个timeout时间来让它自动fail。

      每个你处理的tuple， 必须被ack或者fail。因为storm追踪每个tuple要占用内存。所以如果你不ack/fail每一个tuple， 那么最终你会看到OutOfMemory错误。

       <pre name="code" class="java"> publicclass SplitSentence implements IBasicBolt {
        public void prepare(Map conf,
                            TopologyContext context) {
           }
 
        public void execute(Tuple tuple,
                            BasicOutputCollector collector) {
                  String sentence = tuple.getString(0);
                  for(String word: sentence.split(" ")) {
                    collector.emit(newValues(word));
                  }
       }
 
        publicvoid cleanup() {}
 
        publicvoid declareOutputFields(
                        OutputFieldsDeclarer declarer) {
            declarer.declare(newFields("word"));
        }
    }