基于 Kafka的Hyperledger Fabric Ordering 服务

为了理解 Hyperledger Fabric 基于 Kafka 的共识机制，翻译了一下 Kostas Christidis 的 《一个基于 Kafka 的 Fabric Ordering 服务》～

_________________

A Kafka-based Ordering Service for Fabric

作者：Kostas Christidis

1、介绍

我们使用 Kafka 来提供 Ordering 以及支持区块链崩溃容错。如下图所示，一个 Ordering 服务由一个 Kafka 集群和与之配套的 ZooKeeper、一些部署在 Kafka 集群和客户系统之间 的 Ordering 服务节点（Ordering Service Node，简称 OSN）构成。多个 OSN 之间不能直接通信，而是通过 Kafka 交互。

这些 Ordering 服务节点 OSN 需要做 1）客户端授权验证；2）提供接口给客户系统读写一个链（这里的"链"代表一组客户（例如一个 "Channel"）可以访问的日志记录（log））；3）当新建或重新配置一个链时，根据配置过滤交易并校验合法性。

Kafka 从何而来？

Kafka 中的交易记录被存在一个 topic 分区（topic partition）内。一个 Kafka 集群可以包含多个 topic，每一个 topic 可以有多个分区，每个分区是一个排好序的、不可变的、可逐步追加的交易记录序列。如下图所示，一个 topic 包含三个分区，分区中的每条记录都有一个 offset 索引值。

方案 0.假设每一个链我们有一个独立的分区。一旦 OSN 执行完客户授权验证以及交易过滤后，他们将把正在处理中的归属于某个链的交易拼接到这个链对应的分区内。然后他们也能消费这个分区，获得一个排好序的、与其他 Ordering 服务节点相同的交易列表。

图三可以从总体来帮助我们理解用 Kafka 怎么实现 Ordering。在这里，我们假设所有客户交易均属于同样的链。OSN们将处理的客户交易追加到 Kafka 集群中的同一个分区。他们可以从该分区读取这些交易。网络中的这些交易对于所有的 OSNs 而言均相同。

这样就完了吗？我们继续往下看。

在这个例子中，每一个客户交易都是一个独立的区块（以下简称“块”）。而实际上，OSN 需要将收到的每一个交易放到一个“区块”内，用 Kafka 集群分配的 offset 作为这个块的编号，然后加上签名（用于审计）。客户发起的所有 deliver RPCs 均通过建立一个这个分区的 Kafka 消费者形式来访问特定的块或者 offset 编号。

这确实可以工作。

问题 1.但是考虑一下处理的交易并发达到 1K tps 的情况。在这种情况下， Ordering 服务必须在一秒内产生 1K 签名。并且，接收端的客户必须能够在一秒内能够验证 1K 的签名。因为签名和验签一般耗时较长，所以达到这个指标很困难。

方案 1.用批处理来代替一个块仅包含一个交易。例如针对问题 1. 中所描述的问题，假设我们一个批次处理 1K 条交易，那么 Ordering 服务一秒只需计算一次签名，接收端的客户一秒内也只需验证一次签名。

这好多了。

问题 2.但是假设交易流的速率并不均匀了会怎样？假设一个 Ordering 服务刚刚处理一个包含 1K 交易的批次，现在内存中存了 999 个交易，在形成一个新的处理批次前还差一个交易，但是在接下来的一个小时内没有一个交易进来，那么这 999 个交易的处理时间就无端端地延迟了下去。

这显然不太好。

方案 2.我们可以设置一个批处理 timer. 每当积累的交易数达到指定数量 batchSize 或者等待时间超过指定的时间 batchTimout，Ordering 服务就会产生一个块来处理积累的交易。

问题 3.但是，基于时间窗截断的块增加了服务节点之间的协调工作。基于同样的 batchSize 来切割一个消息块很容易，对于所有的 OSN 也可以得到同样的块序列。但是当用时间窗来切割时，问题就变得复杂。例如我们假设 batchTimeout = 1s 并且有两个 OSN。一个批次刚处理完后一笔新的交易进入 OSN1 并且追加在指定分区内。OSN2 在 t=5s 时读取了这个消息，然后设置了新的定时器 t = 6s。而 OSN1 在 t= 5.6s 时读取了这笔交易，然后设置了新的定时器 t = 6.5s (5.6s + 0.0,...,0.9s)。然后第二笔交易来了并被追加在分区内，此时 OSN2 将在 t = 6s 时读取而 OSN1 将在 t = 6.5s 时读取。此时，我们就遇到一种情况，那就是 OSN1 将两个消息都封装在一个块内，但是 OSN2 却只封装了第一笔交易。显然，这两个 OSN 处理的块序列不一致，这不合适。

方案 3.利用一个特别的协调信号来帮助 OSN 基于时间窗截断消息块 。假设每一个 OSN 在切割 块X 给分区发出了 “切割 块X 的时间点”的消息，然后等到分区回复一个“切割 块X 的时间点”（这个时间点可能与 OSN 发出的不同，下文简称 TTC-X）后才真正切割。如果每一个 OSN 均等待 TTC-X 才切割块，那么我们就完成了一次数据对齐。这样情况下，每一个 OSN 切割 块X 要么根据 batchSize 要么根据收到的第一个 TTC-X 消息来处理。这意味着所有的后续针对同样的 X 的 TTC-X 消息均会被所有 OSN 忽略。如下图四所示，OSN 插入交易和 TTC-X 消息到分区内。

现在，我们已经找到了一种方法利用长度或时间窗来切割消息块，并且采取了措施让所有的 OSN 处理的 块序列 一致。

但这就万事大吉了吗？想的美。

问题 4.添加批次后，块编号不再像一个块仅包含一个消息时那样与 Kafka 偏移量 offset number 相同。所以当 Ordering 服务收到一个从 块 5 开始的 Deliver 请求时，5 这个编号对于客户要寻找的消息编号 offset 而言没有概念。

这显然不太好。

方案 4a.一种方法是我们记录下一个 块元数据（block metadata）包含的消息编号区间，例如块4包含的offsets: 63-64。当一个客户想访问块5开始的消息流时，他可以通过在分区内寻找 offset 65 来解决。但是这个方法有两个问题。第一，我们破坏了 Deliver API 以 块编号作为参数的约定；第二，如果一个客户只想随意访问某一个块 X（没有相关的块历史）或者只是想获得最新的块时，他是不知道这个块对应的 offset 编号的。同样的，OSN 也不知道。

方案 4b.所有，每一个 OSN 对于每一个链均需要维护一个表，在块编号和它们包含的第一个消息编号之间做个映射，如表一所示。需要注意的是如果一个 OSN 在映射表内没有找到客户请求的块编号对应的 offet，他将无法处理客户的这个 Deliver 请求。

有了这个查找表，方案 4a 中所提到的块元数据就不需要了，也不需要客户在 Deliver 请求内明确指出块编号对应的消息编号。利用这个查找表，OSN 帮客户找到所需的块编号对应的消息编号。至此，问题 4 可以解决了，只是 OSN 需要维护一些状态。

这确实可行。但还是存在两个问题。

问题 5.每当 OSN 收到一个 Deliver 请求，他都需要根据所需的块编号找到相应的消息编号起始地址，然后打包成块并签名。这个打包和签名每次请求都要做而且重复计算，太浪费了。

问题 6.批处理的方法，使得处理一个 Deliver 请求变得比建立一个 Kafka 消费者、寻址然后得到所有的记录复杂多了。在这个方案里，块编号与消息起始编号的映射表需要每次都用到，增加了额外的开销。

我们解决问题 5，稍后再详细聊聊问题 6。为了克服这个问题，我们必须在创建了块之后持久化它们，只是当使用时我们需要读取整个持久化的块。

方案 5a.为了解决这个问题，我们尝试完全依赖 Kafka 机制。在分区 0 之外，我们新建立一个分区 0 的引用 分区 1。每当 OSN 切割出一个块时，他们就追加在分区 1 内，并且所有的 Deliver 请求都通过 分区 1 来处理。因为每一个 OSN 都将它们创建的块追加在 分区 1 内，所以我们不能指望块序列的顺序与对应的消息序列一样，所以它们可能会重复，也可能会乱序。

如图五所示的分区 1 中，序列从左向右增长，但是追加的块编号顺序确是 3-3-1-2-2-4，不过对于每一个 OSN 而言，他们的顺序是从小到大的。

这意味着对于分区 1，Kafka offset 编号与区块编号也对应不起来，也需要一个区块-消息编号的映射表。只不过分区 1的 offset number 映射到的是一个区块，而不是一个分区 0 内的交易。

这是可行的，只不过需要注意问题 6 “Deliver 请求的处理逻辑很复杂并且映射表每次都需要调用”依然存在。

方案 6.如果你自己思考一下，造成这个问题的根源是新引入的消息不论是分区 0中的TTC-X还是分区 1中的 区块 X都大量冗余，那么我们怎么解决这些冗余消息呢？

方案 6a.首先，先明确一个规则：当你正在处理的消息与你之前从分区内收到的信息一致（不包含签名）时，放弃继续处理。回头再来看看图五，如果 OSN1 已经知道 OSN2 在自己形成区块 3之前就已经形成了区块 3，他将放弃将其追加到分区 1（此处的描述同样适用于图四中的分区 0）。这种方法可以避免一些冗余消息但是并不能完全去除，因为还是有一些消息几乎同时追加到分区。尽管你可以通过类似于增加延时的方法来错开每个 OSN 的消息追加，但这个问题无法避免。

方案 6b.如果我们有一个 leader OSN 负责向分区 1内追加区块会怎样？有很多方法选举出一个 leader。例如你可以让所有的 OSN 竞争一个 zooKeeper 的 znode 或者指定第一个追加 TTC-X 到分区的 OSN 为 leader。另外一个有趣的方法是让所有的 OSN 都属于一个 Kafka 消费者组，组内的每一个 OSN 均排他性的掌管一个 topic 内的分区。（假设一个 topic 包含所有链的 分区 0，另一个 topic 包含所有链的 分区 1。）这样，负责追加 TTC-X 消息到分区 0 或者 区块到分区 1 的 OSN 将是那个链分区 0 的 owner。这个状态可以由 Kafka 来管理并且可以通过 Kafka Metadata API 来查询。在这种情况下，当一个 OSN 接收到一个不归属于他掌管的 Deliver 请求时，他将会把它重定向到相应的分区 owner 那里去。

这是个方案，但是如果这个 leader 发出区块 X 到分区 1后就崩溃了，这个消息还在送往分区 1的路上。此时，其他 OSN 通过 zooKeeper znode 机制发现这个 leader 奔溃了，于是将选举新的 leader。新的 leader 发现自己的缓冲区内有 块X 还没有被存入分区，于是他将执行追加操作。但是，老的 leader 送出的 块X 最终也将追加到分区内。显然，两者重复了。如图六所示。

方案 6c.日志压缩（log compaction）可能会有用。日志压缩使得 Kafka 在每个分区内为每一个 message key 仅保留最新的 value。（Kafka内所有的记录均是 key/value 结构），如图七所示。

所以，如果所有的 块 X 有同样的 key 并且不同的 X 有不同的 Key，如果我们打开日志压缩开关，我们就可以去掉所有的重复消息了。但是因为日志压缩使用了最新的 key/value，所以 OSN 可能会依然在使用过时的查找表。图七中，假设 key 代表了块映射列表，一个 OSN 只处理了两条消息所以他的查找表仅包括映射 ，。同时，这个分区又被压缩成图中下方的表格，其中，的映射均被压缩掉，所以如果查询这两个映射关系就会报错。还有一个问题也很重要。那就是，分区 1 内的区块将不会以升序形式排列，如图七中的序列是1-3-4-5-2-6。所以，Deliver 逻辑依然很复杂。考虑到映射表可能会过时的问题，显然我们不能用日志压缩。

所以，目前没有办法解决问题 6。既然如此，我们来看看是否可以找到更快的方法解决问题 5，也就是持久化区块并且检索效率更高。

方案 5b. 与方案 6a 不同，这里我们不引入另一个 Kafka 分区，依然只采用包含所有交易和 TTC-X 消息的分区 0。但是我们在每一个 OSN 处为每一个链保留一个本地日志（local log），如图八所示。

这样做还有一些好处，如下：

1、他可以提供解决问题 6的方案 6d，即通过序列化读本地账本就可以响应一个 Deliver 请求。（因为 OSN 在写本地日志可以做过滤，所以没有重复。也不需要映射表。只不过 OSN 需要维护他读的最后一个 offset number 便于后续在此基础上继续从分区内读取数据）。

2、他最大化地重用了 orderer 基础性代码，且 Deliver 方法也与所有区块链实现保持了一致。

当然，采用本地账本来处理 Deliver 请求也存在一些缺点，例如，他比直接从 Kafka 内读取要慢。不过这个也没办法，因为我们毕竟需要 OSN 来处理一些事情。

尤其是 replay 请求，其他一些方案（方案 4b 和 5a），不论是 4b 中的打包，还是 4b 和 5a 中的映射表查找，都需要 OSN 来处理。所以，在没有本地账本前，在 OSN 这里就要牺牲一些效率。

对于 Deliver 请求，例如 keep current(类似于 tail -f)，相对于 方案 4b 而言，6d 额外付出的代价是还需要存储区块到本地账本然后在查找他。在方案 5a 中，这个区块需要到分区 1内兜一圈，依赖于部署环境，他的速度未必比 6d 快。

综上，一个兼顾了性能和复杂性的 Ordering 服务，可以由以下几部分组成（为每一个链）：

1、使用 Kafka 的一个分区来存放客户交易和 TTC-X 消息；

2、在每一个 OSN 本地维护一个账本，用来存储产生的区块。