前言

好久没有写文章了，然后一连就写了三篇，
前两篇文章
Storm入门（一）：编程模型
 Storm入门（二）：架构模型和集群部署
都是一些比较简单的入门教程，这一篇我们来聊一聊稍微高级点的话题，
关于 Storm 的 ACK 机制。

不过话又说回来，其实大数据领域，该机制还是显得比较鸡肋的！！！

ACK机制有什么用？

我们知道 Storm 是一个常驻服务，消息源源不断的来，他源源不断的处理，那肯定在有些情况下会导致消息的不正确处理，比如worker进程挂掉了，那么正在被处理的消息很可能就会丢失掉，那么该如何解决这个问题呢？这时候我们就可以引入 ACK 机制了，当消息没有被正确处理时，可以通过 ACK机制重新发送该消息进行处理。

当然，大多数时候，一条两条数据的异常，并不在我们的考虑范围内，所以并不是所有任务都要引入 ACK 机制

开启 ACK 机制

首先我们来看看如何开启 ACK 机制：

spout 发送 tuple 的时候需要指定该消息的 messageId
SpoutOutputCollector.emit(List<Object> tuple, Object messageId)
spout要重写BaseRichSpout的fail和ack方法

 static class MySpout extends BaseRichSpout {
        @Override
        public void fail(Object msgId) {
          //消息失败的时候回回调到这个方法
        }

        @Override
        public void ack(Object msgId) {
          //消息成功执行的时候回回调到这个方法
        }
    }

你可能已经发现这两个回调的参数是 msgId，而不是你发送的 message，所以这个时候需要我们自己在发送数据的时候维护一个缓存，在 ack 回调里面移除，在 fail 里面重发。

Bolt 发送消息的时候需要将原消息当做 anchor 发送
OutputCollector.emit(Tuple anchor, List<Object> tuple)
设置acker数至少大于0：
Config.setNumAckers(1);

通过上面三步我们就可以开启 ACK机制了，不过这里需要注意的是：
如果启用 ACK机制，我们可以采用 IBasicBolt 接口下的 BaseBasicBolt 而不是 IRichBolt接口下的BaseRichBolt，
该 Bolt 会自动进行 ACK 的发送和 anchor的关联，这样就省得我们忘记添加ACK，使得ACK无法正确运行

上面介绍了如何开启一个 ACK，实际上我们也看到了，ACK机制的控制是精确到了 message 的，比如我们Spout 发送这个 message 的时候不指定其 messageId，那么这个message 的数据流就不会被 ACK,

ACK 原理

知道如何使用ACK了，那么我们现在一起来看下 ACK的是怎么实现的吧！！！
1. 首先我们在开启ACK机制的时候设置了 Config.setNumAckers(1),其实就是开启了一个 ACKER 的task，
  ACKER 会负责跟踪 spout 发出的每一个数据流
2. 我们在 Bolt 中的每次 ack()调用就是向 ACKER 汇报本次执行任务结果，ACKER 接受到数据之后会判断该任务是否执行完毕
3. 如果执行完毕，ACKER 会向Spout 汇报说该消息执行完毕，这个时候 Spout就会通过我们重写的 ack 方法让我们处理成功的逻辑
4. 如果执行失败，或者很长一段时间 (Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS可以设置这个时间还) 没有收到 ACKER 的汇报，那么Spout就认为该任务失败了，会通过fail方法告诉我们处理失败的逻辑
看完这个原理，我想你大概也知道了这个 ACK机制其实并无法保证数据有且只有一次被消费，他只能保证你的数据至少被消费一次，很可能会重复消费。
这里还有一个问题就是 ACKER 是如何判断一个消息被处理完成的呢？这里 Storm 使用了一个比较巧妙的方法，下面我们就来说一下这个判断的机制
1. 首先我们需要知道一点：任意值异或同一个值两次，还是其本身，比如 10^3^3 还是 10
2. ACKER 实际上会为每个 message 生成一个ack-val，初始值为 0。(注意这个message，这也是个坑，一个消息进来后，可能产生多个数据流，当某个数据流失败后，却要重试整个message所有的流)，每执行一个 task，即spout或者 bolt 每次执行完毕，会向 ACKER 汇报其 输入Tuple 和 生成Tuple，ACKER会将其本身的ack-val 异或 输入Tuple的id 再异或 生成Tuple的id 生成一个新的 ack-val
3. 下面为了方便理解，我们举个小例子如下图：
  
  image.png
- 假设我们现在有一个message 发送过来了，通过 Spout 生成A1 和 A2 两个touple 分别发送给 BoltA1，BoltB1.最后BoltA1 和 BoltB1将数据汇聚到 bolt。
  那么这个 ACKER 的判断过程大概就是这样：
  1. spout 发送到 A1 和 B1 的时候：ack_val =0^A1^B1
  2. A1 通过 boltA1 生成 A2，ack_val = ack_val ^(A1^A2)
  3. B1 通过 boltB1 生成 B2，ack_val = ack_val ^(B1^B2)
  4. 最后 B2 和 A2 汇聚到 bolt 的时候， ack_val = ack_val ^(B2^A2)
  最后我们把ack_val值的变化都写在一个公式，大概就是：0^A1^B1^(A1^A2)^(B1^B2)^(B2^A2),因为异或的性质，不难得出最后的结果为 0 ，而当ack_val 重新变为0，ACKER就认为你这个数据已经处理完成了
  - 你可能需要注意的问题：
    1. 关于重试机制的控制，当一个消息处理失败，你可以重发它，它失败十次，你还可以重发，但是百次千次呢？所以每个消息最好设置下最大重发次数。哈~~这个要你自己实现哈....Storm 没有提供相关的配置
    2. 消息重复问题，比如一个message有产生了两个数据流，其中一个成功，另外一个失败，当消息重发的时候，成功的那个数据可能就被计算了两次，所以对于结果数据如果有必要，需要进行相关的去重逻辑。至于这个去重，可以好好利用下 messageId。
    3. 如果你有补充，欢迎指教