告别传统金融消息架构：Apache Pulsar 在平安证券的实践

在金融场景中，伴随着业务的扩展，应用系统也相应地增加更多的场景，这些新场景对消息系统提出更多样的需求，导致原有架构面临一系列挑战。在尝试使用 Apache Pulsar 后，平安证券决定在生产环境中进行实践。本文介绍了平安证券选择 Apache Pulsar 的原因，使用 Apache Pulsar 的场景，Apache Pulsar 实践应用中遇到的问题，以及使用 Apache Pulsar 的未来规划。

背景介绍

传统金融公司或券商一般会使用统一接入服务或组件来处理对外业务。接收到用户请求后，根据相应的业务规则将请求转到对应业务系统 / 模块。有些请求会转发给消息队列，请求写入后，下游业务系统从消息队列中获取请求，并在处理后通过消息队列原路返回给客户，整个请求过程封闭运行，功能有限。

消息队列下传统架构带来的挑战

平安证券采用的就是上述的传统架构，目前只支持消息队列。虽然我们有一定的开发能力，但也难以获取到该消息队列的细节信息。同时，由于是定制开发的系统，支持的语言比较有限。现有的消息队列对业务发展和业务创新等有以下不足：

1. 黑盒系统，难以观测：消息队列是一个黑盒系统，我们难以观测到架构的细节；

2. 直接交换（Direct Exchange），无法路由：由于架构目前只支持消息队列，无法支持需要路由的场景；

3. 弱校验接入，安全风险高：现有系统的密码认证、校验等检验较弱，安全风险较高；

4. 定制系统，有限语言支持：定制系统接入语言的支持有限，导致我们选择范围少，难以在原有系统基础上进行改革。

随着业务扩展和架构改进，公司现有的消息队列系统 / 组件面临着一系列挑战，而系统存在的许多问题如安全性等在金融场景中刻不容缓。

金融场景的业务需求

我们的业务需求主要分为三类：身份识别 & 安全控制、路由分发、审计。

身份识别 & 安全控制

身份识别主要用于确定接入消息队列的客户端和接入者的身份信息，指定相应的安全规则，拒绝不合法接入者，进而实现预期的安全要求。从最基础的层面看，需要识别控制接入的系统、IP，根据业务场景及特定需求，进行权限限制。

路由分发

路由分发指消息根据相应的规则由写入队列路由至对应的队列。现有的消息队列支持的场景有限，若想要支持更多场景，需要投入大量的时间和精力来进行开发（涉及上下游系统的改造），同时会引入其他问题。较好的解决方案是消息队列系统原生支持更多模式及特性，比如 TOPIC 模式、流式消息处理。如果消息队列系统可以支持路由，那么系统的接入复杂度就会大大降低，可以通过更优的方式对接入层进行操作，每个系统只需要对接一组 topic，路由负责分发；也可以更有针对性地优化性能（路由、转发、协议转化都是消耗性能的操作）。

原有系统架构通讯机制是点对点，封闭运行，请求消息无法共享，只能间接采用适配器或日志采集方式实现分发，此类做法难以有效满足实时性要求。

审计

消息的发布者 / 接收者都属于整个系统的参与者，并且是重中之重。系统安全性的主要影响因素就是系统的所有参与者；因此，从安全角度出发，对消息的审计要求相对较高。另一项比较急迫的需求是对消息流向进行控制。如果可以进行身份识别和安全控制，则可以在审计时完善和优化安全信息，进而保证在业务入口处拒绝无效、非法请求，保证内部系统健壮。此外，记录接入的消息发布者 / 接收者的信息还可以用于异常情况监控、稽核审计。

新增业务的系统需求

新增业务对消息系统提出了更高要求，主要包括可用性、消息发送延迟、扩缩容、消息回溯等。

需求一：高可用、低延迟

对于互联网行业而言，高可用低延迟是系统的基本要求。从单点到灾备，到同城跨机房，再到异城跨多中心，或者是先跨城、灾备，再跨城多中心（两地三中心）的模式都已经越来越常态化，很多公司的业务系统正在或将会往这方向发展。这样的系统对高可用、低延迟的要求比较高。因此需要考虑当系统复杂度增加（如灾备、跨城等场景）时，如何将延迟降到最低。

需求二：快速扩容与恢复

对于金融业而言，业务的主要特性之一是请求可能会在某个时间段或某个周期激增，过了这个时间窗口后，流量逐渐恢复正常。这一特性要求系统可以快速横向扩缩容，出于成本考虑，按照最高流量部署整个系统架构显然不合理。最好的解决方案是系统可以根据单层流量合理安排系统架构或系统部署方式，在流量突然增加时，系统可以快速扩容，支撑业务。最理想的情况是系统的所有组件都有快速扩缩容、恢复能力。

需求三：消息有序、消息防重

在一些特殊业务场景中，需要保证消息有序或防重。我们经常对一些接口进行幂等操作，如果可以保证上游消息不重复，就可以减小下游的压力。

需求四：可回溯、序列化

如果业务系统出现问题，但在测试环境中难以复现这一问题，就需要引入消息回溯。消息回溯指重放一遍出现问题的时间窗口中的所有请求，验证是否能复现问题，并排查问题，这样可以大大减轻排查问题的工作量。此外，我们还可以借助这一功能进行灰度验证和并行验证。

选择 Apache Pulsar

基于上述业务需求和系统需求，发现 Apache Pulsar 的诸多特性完美契合了我们的需求。

1. 集群模式，支持跨集群同步。建设系统双活，跨集群的地域复制在客户端无感的情况下实现消息同步。

2. 计算存储分离。根据使用情况横向扩展存储 / 计算，客户端对此操作无感知。基于二级存储的功能，扩展消息的使用场景，为数据分析、消息审计提供可能。

3. 客户端接入认证模块插件化，支持自定义开发。因业务需求，在客户端接入时，要求鉴权、认证，有效保证消息的来源可靠、可控。

4. 完善的 Rest API，可查看队列情况。之前使用的消息系统有很好的性能，但在可观测性方面有所欠缺，给系统排障造成困难，同时消息系统的管理方式较为原始，难以适配大规模系统管理的要求。而 Apache Pulsar 完善的 Rest API 不仅可以获取系统运行指标，且有助于集群的高效管理。

5. 基于 Functions 可实现消息的路由开发、过滤和统计等。

6. 可设置消息的持久化模式和过期时间，允许消息重放。

7. 多语言支持，快速便捷接入。

Apache Pulsar 在平安证券的业务场景

平安证券使用 Apache Pulsar 构建统一消息平台，期望整合客户、交易、行情、资金四大数据流，应用于行情分发、实时风控等。本文主要介绍如何将 Apache Pulsar 应用于三个业务场景：请求路由、数据广播和消息通知，新架构的优势和不足，以及其对开发、运维团队的影响。

场景一：请求路由——简化系统

我们的消息路由流程如下图。从 A 组件发出的请求写入到 Topic A，然后由路由模块将 topic 中的信息进行路由，分发到多个对应的 topic，订阅了这些 topic 的下游组件就可以处理相关的消息。组件 A 只需要向固定的队列写入消息，不需要关注 Topic B、C、D 的信息，下游系统只需要了解接收消息的队列，不需要关注 Topic A，从而简化整个网络的结构。

这种消息路由模式简化了系统的整体架构，目前我们的路由系统仍待优化：

1. 虽然路由分发的工作量得以减轻，但排查问题的步骤有所增加。比如在组件 A 发送消息后，组件 B 没有收到消息时，需要先检查组件 A 是否写入消息到 Topic A、路由模块是否成功路由这一消息，再看组件 B 是否正确订阅了这条消息。

2. 从目前的测试效果看，由于消息链路变长，时延增加。

3. 由于每个队列的消息都会持久化，导致存储和队列中都出现数据冗余。

4. 路由模块是新增模块，运维的学习成本较高。

场景二：数据广播——降低时延

数据广播是我们使用 Apache Pulsar 的另一个业务场景。数据广播采用发送 / 订阅模式，主要用于同步消息。很久之前，我们不需要同步行情到业务系统，或是通过其他方式（如同步数据库）实现。但随着业务的增长，同步时效和用户体验的竞争度越来越激烈。如何可以让用户更快地看到信息？以同步行情的场景为例，先同步数据库再查阅的方式，时延相对较长；而在广播模式中，业务系统只需订阅所有需要的 Topic，查阅时即可直接读取数据，有效降低时延。

场景三：消息通知——安全管控

我们使用到 Apache Pulsar 的第三个场景是消息通知。虽然消息通知涉及到的业务相对较少，但这一业务场景十分重要。整体业务流程图如下。由于信号源不唯一，因此在消息发布到计算引擎后，计算引擎需要根据信号源的信息进行逻辑、安全等方面的计算。计算完成后调起 Task，再由激活的 Task 向相关业务系统发送业务请求，执行后将结果返回给发起信号源的服务，该服务根据返回的结果触发下一个信号源。

这一场景涉及到的业务对安全和管控的要求非常严格，不仅需要限制信号源发送的消息或信号，截断 / 过滤某些信号，还需要对返回的结果进行处理：哪些可以返回，哪些需要过滤掉或转换成其他内容。如果不使用消息队列方式，消息源会直接发送消息给计算引擎，在计算引擎执行安全或管控策略后，将消息发送到 Task; 在 Task 执行完成后，其结果需要再进行一轮安全管控处理。这一部分的重复操作对性能影响较大，同时策略更新、信号状态查看的时效性没有那么实时。

引入 Apache Pulsar 后，我们将管控审计模块剥离出来，专门针对信号队列和结果队列进行过滤、审计、统计等操作，并实时输出结果到管理端。运维或审计人员在看到这些信息后，可以控制、更新相应策略。这一模式不仅可以精简数据流，还可以增加数据补充渠道，也更清晰地定义了各服务模块的边界。

问题发现与解决方案

目前我们主要在上述三个场景中探究了 Apache Pulsar 的使用，并逐步上线生产。在使用过程中，我们发现了几个问题，并在此分享我们的解决方案以供参考。

1. 实现 REQ-REP 模式

我们遇到的第一个问题是如何实现请求 - 响应（REQ-REP）模式，我们的解决方案是通过总线模式进行兼容。

目前常见的调用方式是客户端发起调用请求，服务端处理完成后返回响应即可。但引入总线以后（同步转异步），在多节点的部署场景中，节点 1 发出请求，服务端收到请求后返回处理结果，所有节点都需要监听这条处理结果，节点 2 收到归属节点 1 的响应消息时应该如何处理？节点 2 需要先订阅并获取回包的消息，判断是不是自身节点发起请求的响应，如果不是，则丢弃此消息。假如按照这种模式进行实现，则在发送消息时，每个节点都需要缓存自身发送的消息 ID；服务端处理完以后，按照协议回包数据需要带上请求的消息 ID，每个节点都订阅获取所有回包，并校验缓存中是否有该消息 ID，若不存在，则丢弃消息。

该实现方式下存在一个非常严峻的问题亟待解决：节点发起一个查询大量数据的请求时，假定 Apache Pulsar 设置一个消息 的大小为 8M，TPS 为 1000，那是不是每个节点都要收到这么多请求的回包流量呢？假如有 5 个节点，每个节点本应该只接收 200 个请求的回包流量就够了，但现在的模式需要每个节点承受 1000 个请求的回包流量，而其目的仅仅是为了过滤操作。如果节点负载性能达到上限，需要扩容节点，将导致网络带宽成倍增加。由于 Apache Pulsar 可以支持大量 Topic，虽然通过给每个节点配置一个回包队列等方式可以解决这一问题，但我们想尝试通过 broker 的 FILTER 功能，来解决该问题。

2. 实现读写分离

消息广播场景会涉及到读写分离。如果增加大量订阅节点，最好避免将所有节点的链接集中在 Topic 的 owner broker 上。针对这个问题，可行的解决方案是合理分配使用的 Topic 和 Partition。我们目前使用的 Apache Pulsar 2.7.2 还不支持读写分离，计划把 Apache Pulsar 升级到 2.8，就可以轻松实现读写分离，满足消息广播场景的需求。

3. 解决多网卡问题

基于公司网络安全考虑，内部存在多种网络分区及网段，不同的网络分区 / 网段使用不同的 IP，服务器存在多个网卡，供跨分区系统间通信。目前如果使用 IP 注册 broker，只能注册某个网段的 IP；如果使用域名注册 broker，则不同网络区域的 DNS 解析又需要进行不同的配置。如果 broker 可以支持多网卡通信，这些问题就不存在了。目前我们的解决方案是用 proxy 代理客户端的请求，外部系统也只连接到 proxy，我们也会为 proxy 增加一些高可用的配置。

未来规划

目前 ，我们在单机房、单集群线上小规模运行 Apache Pulsar，在上线初期未考虑建设双活。作为业务系统的基础设施，Apache Pulsar 自身可用性极为重要。因此，我们计划基于同城双中心单集群建设进行双活规划，如图如示：

在测试和使用 Apache Pulsar 的过程中，我们遇到了一些问题，感谢 Apache Pulsar 社区的积极响应。我们期待更多地参与到 Apache Pulsar 的研发中，也期待为 Apache Pulsar 和 Apache Pulsar 社区做出贡献。

作者简介：

王东松，平安证券经纪业务事业部研发工程师。

陈翔，平安证券经纪事业部架构师。