译者简介

原文由李鹏辉公布于 StreamNative 英文站点：https://streamnative.io/en/bl…

译者：王嘉凌 @中国移动云能力核心，挪动云 Pulsar 产品负责人，Apache Pulsar Contributor，沉闷于 Apache Pulsar 等开源我的项目和社区

Apache Pulsar 社区在刚刚公布的 Pulsar 2.8.0 版本中实现了一个里程碑式性能：Exactly-once（准确一次）语义。在这之前，咱们只能通过在 Broker 端开启音讯去重来保障单个 Topic 上的 Exactly-once 语义。随着 Pulsar 2.8.0 的公布，利用事务 API 能够在跨 Topic 的场景下保障音讯生产和确认的原子性操作。接下来，我会解释一下这两种形式的含意和实现办法，以及在实时数据音讯和流计算零碎中如何应用 Pulsar 事务个性来实现 Exactly-once 语义。

在深刻了解 Pulsar 事务个性之前，咱们先来回顾一下音讯语义的概念。

什么是 Exactly-once (准确一次)语义?

在分布式系统中，任何节点都有可能出现异常甚至宕机。在 Apache Pulsar 中也一样，当 Producer 在生产音讯时，可能会产生 Broker 或 Bookie 宕机不可用，或者网络忽然中断的异常情况。依据在产生异样时 Producer 解决音讯的形式，零碎能够具备以下三种音讯语义。

At-least-once (至多一次)语义

Producer 通过接管 Broker 的 ACK（音讯确认）告诉来确保音讯胜利写入 Pulsar Topic。然而，当 Producer 接管 ACK 告诉超时，或者收到 Broker 出错信息时，会尝试从新发送音讯。如果 Broker 正好在胜利把音讯写入到 Topic，但还没有给 Producer 发送 ACK 时宕机，Producer 从新发送的音讯会被再次写入到 Topic，最终导致音讯被反复散发至 Consumer。

At-most-once (最多一次)语义

当 Producer 在接管 ACK 超时，或者收到 Broker 出错信息时不重发消息，那就有可能导致这条音讯失落，没有写入到 Topic 中，也不会被 Consumer 生产到。在某些场景下，为了防止产生反复生产，咱们能够答应音讯失落的产生。

Exactly-once (准确一次)语义

Exactly-once 语义保障了即便 Producer 屡次发送同一条音讯到服务端，服务端也仅仅会记录一次。Exactly-once 语义是最牢靠的，同时也是最难了解的。Exactly-once 语义须要音讯队列服务端，音讯生产端和生产端利用三者的协同能力实现。比方，当生产端利用胜利生产并且 ACK 了一条音讯之后，又把生产位点回滚到之前的一个音讯 ID，那么从那个音讯 ID 往后的所有音讯都会被生产端利用从新生产到。

实现 Exactly-once 语义的难点

在分布式消息中间件零碎中实现 Exactly-once 语义面临不少挑战。上面通过一个简略的例子来形容。

假如有一个 Producer 发送一条内容为“Hello StreamNative”的音讯到 Pulsar 上“Greetings”这个 Topic 中，而后有一个 Consumer 会从这个 Topic 中接管音讯并打印进去。在现实的状况下，没有异样呈现，“Hello StreamNative”这条音讯只会往“Greetings”这个 Topic 中写入一次，而后 Consumer 会接管到这条音讯并进行解决，而后通过 ACK 来告诉 Pulsar 音讯已处理完毕。之后哪怕 Consumer 呈现宕机或者重启，也不会再次接管到这条音讯。

然而，异样和谬误往往无处不在。

Bookie 可能呈现宕机

Pulsar 利用 BookKeeper 来存储音讯。BookKeeper 是一个高可用的长久化日志存储系统，写入 Ledger（Pulsar 中 Topic 的一个分片）的数据会保留在 N 个 Bookie 节点上，也就是说，BookKeeper 能够容忍 N-1 个 Bookie 节点的宕机。只有至多有一个 Bookie 节点可用，这个 Ledger 上的数据就不会失落。依靠 Zab 协定和 Paxos 算法，BookKeeper 的正本协定能够保障一旦数据胜利写入 Bookie 中，这些数据将主动复制到属于同一组的 Bookie 节点上永恒保留。

Broker 可能呈现宕机，或者与 Producer 之间的网络呈现中断

Producer 通过接管 Broker 的 ACK 告诉来确保音讯发送胜利。然而，没有接管到 ACK 告诉并不总是意味着发送音讯失败。Broker 可能会在胜利把音讯写入到 Topic 之后，还没有给 Producer 发送 ACK 的时候出现异常，也可能会在把音讯写入到 Topic 之前就出现异常。因为无奈晓得 Broker 出现异常的起因，Producer 在接管 ACK 失败的状况下默认会认为音讯没有发送胜利并从新发送。这意味着在某些状况下，Pulsar 会写入反复的音讯，从而导致 Consumer 反复生产。

Pulsar 客户端可能呈现宕机

实现 Exactly-once 时必须思考到 Pulsar 客户端不可用的状况。很难去精确的辨别客户端是不可复原的宕机了还是只是临时的不可用，但对于 Broker 来说具备这种判断能力是很重要的。Pulsar Broker 须要屏蔽掉非正常状态下的客户端发过来的音讯。一旦客户端重新启动，客户端能够晓得之前产生失败时的状态，并从失当的中央接着解决后续的音讯。

Pulsar 社区通过两个阶段来实现 Exactly-once 语义。在 Pulsar 1.20.0-incubating 版本中咱们通过幂等性 Producer 来保障单个 Topic 上的 Exactly-once 语义。在最新公布的 Pulsar 2.8.0 版本中咱们通过引入事务 API 来保障跨 Topic 场景下音讯的原子性操作。

幂等性 Producer：实现单个 Topic 的 Exactly-once 语义

咱们从在 Pulsar 1.20.0-incubating 版本中通过幂等性 Producer 来保障单个 Topic 上的 Exactly-once 语义开始讲起。

什么是幂等性 Producer？幂等性就是指对于同一操作发动的一次或者屡次申请的后果是统一的，不会因为屡次操作而产生不同的后果。如果在集群（Cluster）或者 命名空间（Namespace）级别开启音讯去重，同时在音讯生产端配置幂等性 Producer，那么当呈现因为异样导致 Producer 重发消息时，反复的音讯只会在 Broker 中写入一次。

通过这个性能能够实现在单个 Topic 下不会有音讯失落，不会有反复音讯，所有的音讯都是有序的。咱们能够通过以下配置来开启这个性能：

• 在 Cluster 级别（针对所有 Namespace 下的 Topic 无效），Namespace 级别（针对该 Namespace 下的 Topic 无效）或者 Topic 级别（针对单个 Topic 无效）开启音讯去重
• 为 Producer 设置任意的名称并且设置音讯超时工夫为 0

这个性能是如何实现的？简略来讲，和 TCP 协定的音讯去重机制十分相似：每条发送给 Pulsar 的音讯都会带有一个惟一的序列号，Pulsar Broker 利用这个序列号来判断和去除反复的音讯。不同的是，TCP 协定只能保障实时连贯中的音讯去重，而 Pulsar 会把音讯体中的序列号保留到 Topic 中，并且记录最新接管到的序列号。所以哪怕 Broker 节点出现异常宕机了，另一个从新接管解决该 Topic 的 Broker 节点也能够判断音讯是否反复。这个原理非常简单，和非幂等性 Producer 相比减少的性能损耗简直能够忽略不计。

Pulsar 1.20.0-incubating 当前的版本都反对这个性能，能够从这里找到这个性能的介绍。

然而，幂等性 Producer 只能在特定的场景保障 Exactly-once 语义，在其余的场景却无能为力。比方：当 Producer 须要确保一条音讯同时发送到多个 Topic 时，负责解决其中某些 Topic 的 Broker 宕机了。如果 Producer 不重发消息，就会导致一部分 Topic 中的音讯失落。如果 Producer 重发消息，就会导致其余 Topic 中的音讯反复写入。

在生产端，Consumer 向 Broker 收回的 ACK 申请属于 Best-effort（尽力服务），也就是说 ACK 申请可能会失落，并且 Consumer 无奈晓得 Broker 是否失常收到了 ACK 申请，在产生 ACK 申请失落时也不会从新发送。这也会导致 Consumer 接管到反复的音讯。

事务 API: 实现跨 Topic 音讯生产和确认的原子性操作

为了解决上述问题，咱们通过引入事务 API 来保障在跨 Topic 场景下音讯发送和确认的原子性操作。通过这个性能，Producer 能够确保一条音讯同时发送到多个 Topic，要么这些音讯都发送胜利，在所有 Topic 上都能够被生产，要么所有音讯都不能被生产。这个性能也容许在一个事务操作中对多个 Topic 上的音讯进行 ACK 确认，从而实现端到端的 Exactly-once 语义。

以下实例代码演示如何应用事务 API：

PulsarClient pulsarClient = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650")
        .enableTransaction(true)
        .build();
Transaction txn = pulsarClient
        .newTransaction()
        .withTransactionTimeout(1, TimeUnit.MINUTES)
        .build()
        .get();
producer.newMessage(txn).value("Hello Pulsar Transaction".getBytes()).send();
Message<byte[]> message = consumer.receive();
consumer.acknowledge(message.getMessageId(), txn);
txn.commit().get();

这段代码展现了如何应用事务 API 实现音讯发送和确认的原子性操作，以及如何应用事务 API 在同一个事务操作中确认音讯。

须要留神的是：

• 同一个 Topic 中能够有局部音讯属于某个事务，局部音讯不属于任何事务。
• Pulsar 客户端中容许存在多个并行的未提交事务。这是和其余反对事务的音讯零碎最基本的区别，能够极大进步事务音讯的解决能力。
• 以后的 Pulsar 事务 API 只反对 READ_COMMITTED（读已提交）隔离级别。Consumer 只会生产到不属于任何事务的音讯和已提交的事务中的音讯，不会生产到未提交和已回滚事务中的音讯。

在 Pulsar 客户端应用事务 API 不须要额定的配置和依赖。

端到端 Exactly-once 流计算变得更简略: Pulsar+Flink 的一个例子

通过 Pulsar 事务 API 咱们已能够在流计算场景中实现 Exactly-once 语义。

在流计算零碎中，有个要害的问题常常被提及：“如果在流计算过程中有一些两头节点宕机了，如何去保障最终计算结果不会出问题？”解决这个问题的要害，在于呈现了异样的节点复原后，如何从产生异样之前的状态从新开始解决流数据。

在 Apache Pulsar 上进行流计算，实质上是对多个 Topic 上的音讯进行 Read-Process-Write 操作。Source 节点从一个或多个输出 Topic 中生产音讯，而后通过 Process 节点对音讯进行一系列的计算和状态解决，最初通过 Sink 节点把处理结果发送到记录后果的 Topic 中。流计算场景下的 Exactly-once 指的是对于 Read-Process-Write 的一整套操作的执行合乎 Exactly-once 语义，即不会失落输出 Topic 上的任何音讯，也不会往记录后果的 Topic 上反复写入音讯。这就是用户在流计算零碎上冀望的 Exactly-once 成果。

咱们来看一个 Pulsar 联合 Flink 进行流计算的示例。

在 Pulsar 2.8.0 之前，Pulsar 联合 Flink 进行流计算只反对 Exactly-once Source Connector 和 At-least-once Sink Connector。这意味着应用 Pulsar 和 Flink 构建的端到端流计算零碎最多只能实现 At-least-once 语义。也就是说发送到记录后果的 Topic 中的音讯可能会反复。

利用在 Pulsar 2.8.0 引入的事务 API，Pulsar-Fink Sink Connector 通过简略的革新就能够反对 Exactly-once 语义。Flink 应用二阶段提交协定（Two-Phase Commit）来保障端到端的 Exactly-once 语义，所以咱们能够实现 TwoPhaseCommitSinkFunction 并嵌入 Pulsar 的事务 API。当 Pulsar-Fink Sink Connector 调用 beginTransaction 时，咱们创立一个 Pulsar 事务并保留事务 ID。后续所有写入到 Sink Connector 的音讯都设置这个事务 ID。当 Connector 调用 preCommit 时将这些音讯写入到 Pulsar。当 Connector 调用 recoverAndCommit 或者 recoverAndAbort 时别离调用 Pulsar 事务 API 来提交或者回滚 Pulsar 事务。这个革新非常简单，只须要在 Connector 中保留 Pulsar 事务 ID 和 Flink Checkpoints 的关联关系，从而在 Flink 的 事务提交和回滚操作中获取到对应的 Pulsar 事务 ID 即可。

基于 Pulsar 事务提供的幂等性和原子性操作，以及 Apache Flink 提供的全局一致性 CheckPoint 查看机制，咱们能够很容易利用 Pulsar 和 Flink 构建出一套合乎端到端 Exactly-once 语义的流计算零碎。

后续

如果你想理解 Exactly-once 实现的更多细节，举荐浏览下 Pulsar 社区改良提案 PIP-31。想理解更多设计细节，也举荐浏览下设计文档。

本文次要是基于用户的角度来介绍 Apache Pulsar 2.8.0 中的新个性事务 API，以及如何应用这个个性去实现 Exactly-once 语义。在下一篇文章中更具体的介绍事务 API 的设计与实现。

近期举办的 Pulsar Summit 北美峰会，有相干演讲《Exactly-Once Made Easy: Transactional Messaging in Apache Pulsar》，可查看视频。

致谢

在过来的一年里有多个 Pulsar Committer 和 Contributor 参加开发了这个里程碑式的性能，在此感激他们：李鹏辉、高冉、丛博、Addison Higham、翟佳、张勇、冉小龙、Matteo Merli、郭斯杰。

同时，再次致谢译者 王嘉凌 @中国移动云能力核心 的优良翻译，让咱们疾速看到了这篇博文的中文版本。

相干浏览

• 技术探索：Apache Pulsar 的事务型事件流

点击链接，获取 Apache Pulsar 硬核干货材料！