作为storm的使用者，有两件事情要做以更好的利用storm的可靠性特征。 首先，在你生成一个新的tuple的时候要通知storm; 其次，完成处理一个tuple之后要通知storm。 这样storm就可以检测整个tuple树有没有完成处理，并且通知源spout处理结果。storm提供了一些简洁的api来做这些事情。

由一个tuple产生一个新的tuple称为： anchoring。你发射一个新tuple的同时也就完成了一次anchoring。看下面这个例子： 这个bolt把一个包含一个句子的tuple分割成每个单词一个tuple。

   

Java代码  

1. public class SplitSentence implements IRichBolt {   

2.             Output Collector _collector;   

3.         

4.             public void prepare(Map conf,   

5.                                 TopologyContext context,   

6.                                 OutputCollector collector) {   

7.                 _collector = collector;   

8.             }   

9.         

10.             public void execute(Tuple tuple) {   

11.                 String sentence = tuple.getString(0);   

12.                 for(String word: sentence.split(" ")) {   

13.                     _collector.emit(tuple,newValues(word));   

14.                 }   

15.                 _collector.ack(tuple);   

16.             }   

17.         

18.             public void cleanup() {   

19.             }   

20.         

21.             public void declareOutputFields(OutputFieldsDeclarer declarer) {   

22.                 declarer.declare(newFields("word"));   

23.             }   

24.         }    

 

 我们通过anchoring来构造这个tuple树，最后一件要做的事情是在你处理完当个tuple的时候告诉storm,  通过OutputCollector类的ack和fail方法来做，如果你回过头来看看SplitSentence的例子， 你可以看到“句子tuple”在所有“单词tuple”被发出之后调用了ack。

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

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

大多数Bolt遵循这样的规律：读取一个tuple；发射一些新的tuple；在execute的结束的时候ack这个tuple。这些Bolt往往是一些过滤器或者简单函数。Storm为这类规律封装了一个BasicBolt类。如果用BasicBolt来做， 上面那个SplitSentence可以改写成这样：

   

Java代码  

1. public class SplitSentence implements IBasicBolt {   

2.             public void prepare(Map conf,   

3.                                 TopologyContext context) {   

4.             }   

5.         

6.             public void execute(Tuple tuple,   

7.                                 BasicOutputCollector collector) {   

8.                 String sentence = tuple.getString(0);   

9.                 for(String word: sentence.split(" ")) {   

10.                     collector.emit(newValues(word));   

11.                 }   

12.             }   

13.         

14.             public void cleanup() {   

15.             }   

16.         

17.             public void declareOutputFields(   

18.                             OutputFieldsDeclarer declarer) {   

19.                 declarer.declare(newFields("word"));   

20.             }   

21.         }    

 
这个实现比之前的实现简单多了， 但是功能上是一样的。

发送到BasicOutputCollector的tuple会自动和输入tuple相关联，而在execute方法结束的时候那个输入tuple会被自动ack的。
我们编写的时候使用IBasicBolt最方便了。或者 extends BaseBasicBolt类

IRichBolt和IBasicBolt对比

IRichBolt和IBasicBolt/BaseBasicBolt对比