Storm容错机制（一）：ACK机制

ACK机制有什么用？

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

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

开启 ACK 机制

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

1. spout 发送 tuple 的时候需要指定该消息的 messageId SpoutOutputCollector.emit(List<Object> tuple, Object messageId)
2. 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 里面重发。

1. Bolt 发送消息的时候需要将原消息当做 anchor 发送 OutputCollector.emit(Tuple anchor, List<Object> tuple)
2. 设置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. 下面为了方便理解，我们举个小例子如下图：

  

  • 假设我们现在有一个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. 如果你有补充，欢迎指教

好吧，ACK的讲解就到这里了，不知道有没有讲清楚，不过实际运用中并没有太大的用处，也可能只是我目前用的不多，基于对一门技术的热情，还是稍稍深入研究了一下，如有不对，欢迎指错