Storm的可靠性与Ack机制

        Storm可以确保由Spout发送的每条消息都会被所有的Bolts完全处理，但是这需要用户来决定是否需要可靠性机制。如果是简单的统计分析，可靠性要求不是很高，则可以选择使用不可靠的Bolts。

       Storm编程中，有各种Bolt，命名格式XXRichBolt或XXBasicBolt。其中，带有Rich的Bolt，是没有可靠性机制的，需要自己手动调用方法collect.ack()或者collect.fail()；而带有Basic的Bolt是封装可靠性机制，在emit一个Tuple时，会在内部自动加入该tuple，进行锚定，所以用户只需要编写处理逻辑，系统会默认调用ack或fail方法。那么，storm是如何来进行ack呢？

注：若要storm系统对一个Tuple进行可靠性跟踪，必须指定一个唯一ID告知系统。因为在Spout中的ack()和fail()方法都有一个参数是messageID，系统是根据该ID来确定成功或失败的。如collector.emit(tuple, tupleID);

先明确一些概念：

1,什么叫做一个消息被所有的Bolts完全处理？

  一个Tuple会经过不同的bolts，形成一个或多个分支，这些分支叫做Tuple树，当Tuple树被完全处理时，这个tuple才被完全处理；当中间某个节点处理失败或超时（超时时间，可以再定义Topology的Config时，进行修改属性conf.setMessageTimeoutSecs），tuple都是处理失败的。

2，锚定（Anchor技术）：即collect.emit（inputTuple，new Values（……））。BasicBolt会在内部自动锚定到输入tuple；RichBolt默认是没有锚定的。

------------------一个tuple的ACK流程如下---------------------

1，Spout初始化，产生一个唯一标识taskID；

2，Spout创建一个Tuple，有个tupleID（64位随机数）；

3，Spout发射Tuple（一定要指定一个唯一ID，来开启跟踪），并同时发送消息到Acker，要求其对Tuple进行跟踪；

4，Tuple经过一系列bolts，产生一个anchor tuple列表；

5，Bolt调用OutputCollector.ack()，进行一系列异或操作，若结果是0，则表示ok，否则fail；

6，根据taskID，回调原始Spout的ack()或fail()方法；

-----------------------------------------------------------------------

注意：方法ack、fail和nextTuple是在同一个线程中完成的，因此应当尽量避免在spout去调用sleep()方法。

如果必须要控制消息的发送速率，可以再开启一个异步线程来读取数据到队列，再由spout去取队列中的数据。

如果不关心数据是否丢失，或者想测试是否有某个bolt拖慢了spout的速度，可以进行如下配置：

1，在定义Topology时，conf.setNumAckers(0);

2，在spout中，不指定唯一的messageId，即不开启跟踪；

3，在bolt中，使用richBolt，即不进行锚定，发射新的tuple；

 

参考网址：找不到了。。。

如有错误，请指出