Twitter Storm源代码分析之CoordinatedBolt

共计 2065 个字符，预计需要花费 6 分钟才能阅读完成。

关于 Twitter Storm 的新特性：Transactional Topology 被问到的最多的问题是: Storm 是怎么知道一个 Bolt 处理完成了它所有的 tuple 的？其实要做到这一点还是有蛮多事情要做的, 幸运的是 Storm 已经提供了一个 Bolt，帮我们把这些事情都做掉了。这个牛逼的 bolt 就是
CoordinatedBolt. 重要的是CoordinatedBolt 的实现也是在 storm 的原语:spout, bolt 这些基础之上的 — 也就是说即使作者不提供，我们自己也可以实现。我们来看看这个类的实现原理。

虽然 CoordinatedBolt 所发挥的作用很牛逼，但是其实它的原理并不是很复杂。它现在被用在两个场景里面:

• DRPC
• Transactional Topology

在看 CoordinatedBolt 的原理之前，我们先看看到底什么叫”处理完了”, 到底处理完什么了?
其实 CoordinatedBolt 对于业务不是完全没有侵入的，要使用 CoordinatedBolt 提供的功能，你必须要保证你的每个 bolt 发送的每个 tuple 的第一个 field 是 request-id，那么所谓的”做完了”的意思是说当前这个 bolt 对于当前这个”request-id”所需要做的工作做完了。这个request-id 在DRPC里面代表一个 DRPC 请求；在 Transactional Topology 里面代表一个 batch.

CoordinatedBolt 的原理是这样的：

• 对于用户在 DRPC, Transactional Topology 里面的 Bolt，都被 CoordinatedBolt 包装了一层：也就是说在 DRPC, Transactional Topology 里面的 topology 里面运行的已经不是用户提供的原始的 Bolt, 而是一堆 CoordinatedBolt, CoordinatedBolt 把这些 Bolt 的事务都代理了。
• 有了这个代理层，CoordinatedBolt 就可以做它的工作了。
• 它会在自己这里维护以下几个数据：
  • 哪些上游 task 要给我发 tuple？（通过构造 topology 的时候所提供的 grouping 信息可以得知）
  • 我给哪些下游 task 发 tuple? (同样通过 grouping 信息可以得知)
• 每个 CoordinatedBolt 在每次真正 bolt 发出一个 tuple 之后，它都会记录下，这个 tuple 发给哪个 task 了。
• 等它所有的 tuple 都发送完了之后（怎么知道发送完了？等会再说，少安毋躁），它通过另外一个特殊的 stream 以 emitDirect 的方式告诉所有它发送过 tuple 的 task，它发送了多少 tuple 给它。
• 一个 bolt 在接到所有的上游 task 发送的 tuple 个数信息之后，对比它接收到的 tuple 数量，如果数量对上了，说明它接收到了所有的 tuple — 它处理完成了。
• 这样它处理完成了，它可以重复上面的步骤通知它的下游，它的下游再通知它的下游的下游等等。
• 总结一下，每个 tuple 怎么知道自己处理完成了的？都是靠它的上游通知的。所以只要一个 bolt 有上游，它就能够知道自己什么时候完成。
• 那总有一个 bolt 是没有上游的 — 最上面那个 bolt。那么这个 bolt 是怎么知道自己处理完成的呢？靠的是 storm 的 ack 系统 — 只要它 ack 了它的上游（某个非 CoordinatedBolt, 在 DRPC 里面就是 PrepareRequest)发送过来的 tuple, 它就完成处理这个 tuple 了。— 也就是说对于最上面那个 bolt 来说它只要处理完一个 tuple(相对于它的下游要处理很多 tuple 才算完成)

具体原理如下图:

正如我们在上面讨论到底什么叫”做完成”的概念的时候，我们说了，CoordinatedBolt的使用对于业务是有侵入的：你必须要在你的每个 tuple 的第一个字段带上当前 request-id, 否则CoordinatedBolt 就跟踪不了了。一种更优雅的方式是网络协议栈里面 IP, TCP 协议的处理方式。IP 包在 TCP 包的外面包上 IP 层需要的信息，而不要求把 IP 层需要的信息掺杂在 TCP 的包字段里面，TCP 层在发送数据的时候只组装 TCP 的那些字段，到了 IP 层自动加上 IP 层的信息。而 IP 层把数据包传给 TCP 层之前也自动去掉 IP 层的那些信息，TCP 只会看到自己层的那些字段，毫无侵入。。作者对于这个问题提了一些改进的措施在这里

推荐阅读：

Twitter Storm 安装配置（集群）笔记 http://www.linuxidc.com/Linux/2013-05/84307.htm

安装 Twitter Storm 集群 http://www.linuxidc.com/Linux/2012-07/66336.htm

Twitter Storm 安装配置（单机版）笔记 http://www.linuxidc.com/Linux/2013-05/84306.htm

Storm 实战及实例讲解一 http://www.linuxidc.com/Linux/2012-08/69146.htm