博文｜Apache Pulsar 在自研数据管道中的技术实践

关于 Apache Pulsar

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。 
GitHub 地址：http://github.com/apache/pulsar/

本文作者为江谋晶，谊品生鲜高级研发工程师。主导数据管道系统的设计与研发，借助 Apache Pulsar 作为数据同步工具，并落地实现增量数据同步的各种应用场景需求。他计划进一步实现数据管道的平台化及可视化，并接入更丰富的数据库类型支持。

背景

数据管道，就是让数据通过一定的传输介质，从一个地点到达另一个地点，从而实现数据的同步或复制，来满足应用需求。随着业务量及数据量的的大幅增长，我们现有的微服务需要再度细化（拆分）。

系统拆分如何做到让用户无感知呢？上线时，通过分流策略将部分用户引流到新的服务中，要求新老系统并行运行一段时间来支撑新服务的试运行到完全落地，从而最大程度上减少生产故障。为了让新服务数据能够与旧系统服务中的数据实时一致，就需要同步数据。随着数据量大幅增长，要加快查询速度，可以将数据复制到 ElasticSearch 中，提高查询速率。

市场上有相关的开源数据同步产品和商业版数据通道工具，不需要人工介入即可实现双边的数据同步复制。但系统重构可能会发生一些表结构或表对象的变动，无法兼容商业的数据同步，需要开发人员介入进行相关处理。我们采用了 Maxwell^[1] + Pulsar^[2] 的自研解决方案：使用 Maxwell 读取 binlog，Pulsar 进行数据传输。Maxwell + Pulsar 实现上层的数据读取，下游业务方实现对应的数据同步逻辑。比如，针对系统重构拆分的数据同步业务场景以及读写分离，将数据复制同步到类似 ElasticSearch 搜索引擎中的业务场景。

为何选择 Pulsar?

在数据管道的系统重构中，我们选择 Apache Pulsar 的原因如下：

•无状态。微服务架构体系中，中间件最好是无状态的。这样启动快，可以随时替换并且可以实现无缝伸缩，弹性扩展。Kafka 不是无状态的，每个 Broker 都包含了分区所有的日志，如果一个 Broker 宕机，并非任意一个 Broker 可以来接管，也不能随意添加 Broker 来分担负载，Broker 之间必须进行状态同步。在 Pulsar 架构中，数据从 Broker 剥离，存储在共享储存内部；上层是上层是无状态的计算层（Broker），复制消息分发和服务（计算），下层则是持久化的存储层（ Bookie ）。所以数据计算和存储相互独立，可以实现数据的独立扩展和快速恢复。•Pulsar 支持流处理和传统的消息队列，大大提升了订阅灵活度。•Pulsar 云原生的架构方便水平弹性扩展，且支持跨地域复制。•Pulsar 支持分区，吞吐高，延迟低。•开源社区活跃，技术支持响应快、服务好。

Pulsar 如何保证分布式消费过程中的顺序

首先，我们了解一下 Pulsar 的订阅模式。Pulsar 有四种订阅模式：exclusive 模式（独占模式）、failover 模式 （故障转移模式）、Shared 模式 （共享模式）以及 Key_Shared 模式。Exclusive 模式只有一个消费者（consumer），接收一个 Topic 所有的消息。

( Pulsar Exclusive 模式消费策略 )

Failover 模式中，同一时刻只有一个有效的消费者，其余的消费者作为备用节点，在主消费者（master consumer）不可用后进行替代（该模式适用于数据量小、解决单点故障的场景）。

Shared 模式中，多个消费者可以连接到同一订阅主题。消息以轮询的方式分布在各个消费者之间，任何给定的消息仅传递给一个消费者。起初我们采用 Shared 模式，因为 Shared 模式具备分布式消费能力，消费速度快。但在生产过程中发现源数据库数据与同步的目标库（ElasticSearch、MySQL）频繁出现数据偏差和数据不一致的问题。经排查发现，是消费顺序错乱导致的，当用户频繁操作某条数据产生了多条 MQ 消息时，Shared 模式下，多个消费者并行消费消息了。

( Pulsar Shared 模式消费策略 )

Pulsar 在 2.4.0 版本基于 Shared 模式推出了 Key_Shared 模式。在 Key_Shared 模式下，多个消费者可以附加到同一订阅。消息在各个使用者之间进行分发，具有相同 key 或相同订购 key 的消息仅投递给一位消费者，不管消息重新发送多少次，它都会被发送到同一使用者。当消费者连接或断开连接时，服务的消费者会更改某些消息 Key（密匙）。Key_Shared 模式保证在 Shared 模式下同一个 Key 的消息会发送到同一个消费者，在并发的同时保证顺序性。

( Pulsar Key_Shared 模式消费策略 )

数据同步场景对消息的顺序要求非常高。当用户不断更新某条数据时，数据库表中对应的记录也在不断更新。数据量大高并发时，需要保证用户变更数据产生的消息顺序与其操作顺序一致，否则会出现同步的该条数据与源数据不一致，产生系统故障。

顺序问题是分布式消费过程中常见的问题。为了保证客户端的有序消费，我们采用 Key_Shared 订阅模式。Key_Shared 模式是 Shared 订阅模式拓展，一个分区可以有几个消费者并行消费消息，但具有相同 key 的消息只路由给一个消费者。其原理是通过哈希来确定目标使用者，每个消费端提供固定范围的哈希值；散列值的整个范围可以覆盖所有的消费端。然后生产消息时指定 key（如下所示），形成闭环，就可以实现有序的存放至指定的分区以及消息有序的消费。具体原理及用法可以参考 Pulsar 官网^[3] 。

key ：{"database":"you_db_name","table":"you_table_name","pk.id":"you_table_Primary key"}

如何过滤重复消息?

消息的传输保障一般有三种：At least once、At most once 和 Exactly once。

•At least once：每条消息会进行多次传输尝试，至少成功一次，即消息可能重复但不会丢失;•At most once：每条消息最多传输一次，消息可能会丢失；•Exactly once：每条消息只传输一次，消息传输既不会丢失也不会重复。

在数据同步场景下，要最大化保证消息的可达性，我们运用 Maxwell 的 At least once 模式，尽可能保证消息传输。在网络不理想时，消息可能已经投递至目标，但接收到超时响应或者未接收成功，Pulsar 会再次投递，从而产生了“重复消息”。

At least once 图解 

为了解决重复消息的问题，我们在数据管道数据链路模型中增加了过滤器，过滤一些重复、无效、重试的消息。

（为了聚焦功能，图中没有显示日志记录、链路追踪等微服务体系中的其他组件）

总结

在需要同步大量增量数据的场景下，我们采用了 Maxwell + Pulsar 的自研解决方案，Pulsar Key_Shared 订阅模式能否很好解决分布式消息消费过程中的顺序问题，在数据管道数据链路中增设过滤器，能保证消息不重不漏。

后续我们计划基于现有的解决方案，充分利用 Pulsar 的特性，将数据管道做成可视化的数据同步中台，接入更多的数据库扩展、完善的监控和日志体系。

相关阅读

引用链接

[1] Maxwell: http://maxwells-daemon.io/
[2] Pulsar: http://pulsar.apache.org/
[3] Pulsar 官网: https://pulsar.apache.org/docs/en/client-libraries-java/#key_shared

---

点击“阅读原文”，获取 Apache Pulsar 硬核干货资料！