【前沿技术】阿里开源搜索引擎 Havenask 的消息系统

2024-02-22
浙江
本文字数：4736 字
阅读完需：约 16 分钟

作者：闻意

Havenask 是阿里巴巴智能引擎事业部自研的开源高性能搜索引擎，深度支持了包括淘宝、天猫、菜鸟、高德、饿了么在内几乎整个阿里的搜索业务。本文针对性介绍了 Havenask 的消息系统--Swift，它是一个设计用于处理大规模的数据流和实时消息传递的高性能、可靠的消息系统。

一、Havenask 介绍

Havenask 是阿里巴巴广泛使用的自研大规模分布式检索系统，是过去十多年阿里在电商领域积累下来的核心竞争力产品，广泛应用在搜推广和大数据检索等典型场景。在 2022 年云栖大会-云计算加速开源创新论坛上完成开源首发，同时作为阿里云开放搜索 OpenSearch 底层搜索引擎，OpenSearch 自 2014 年商业化，目前已有千余家外部客户。

下图展示了 Havenask 中一个完整的搜索服务：在线系统、索引系统、管控系统、扩展插件，且包括了查询流、数据流、控制流。其中数据流的主要载体就是消息系统 Swift，索引系统各个角色之间的数据传递、离线到在线的实时数据传递都是通过 Swift 完成。

Havenask 支持千亿级别数据实时检索、百万 QPS 查询，百万 TPS 高时效性写入保障，毫秒级查询延迟和数据更新，并具有良好的分布式架构、极致的性能优化，能够实现比现有技术方案更低的成本，普惠更多的开发者和企业。

二、Swift 简介

Swift 是一种经过精心构思和设计的消息传递与大规模数据流处理系统，主要目标是为了适应现代分布式架构对弹性、高并发以及低延迟应用的要求。相较于业界广泛应用的传统消息队列技术，Swift 在架构设计理念层面展现了一定程度的创新与突破。

传统消息队列为了确保消息安全性及持久化，普遍采用先将消息落地到本地磁盘再确认发送成功的策略。然而，这种模式无形中对系统的可迁移性、扩展性和资源复用性形成了约束，因为在单一或有限数量的物理机上存储消息数据，一旦发生机器迁移，必然伴随数据迁移过程，且可能引发性能瓶颈。尤其是在与其他应用共享计算资源时，由于 I/O 操作的不稳定性及其对性能的影响显著，往往需要将消息队列运行于专用且独占的硬件环境。

近年来，计算与存储分离架构已经成为学术界与工业界的关注焦点与实践热点。其最大优势在于解耦了计算任务与数据存储的位置限制，使得计算资源在先进调度算法的驱动下能够实现近乎无限的水平扩展。与此同时，诸如 HDFS 和 PANGU 等分布式存储系统可以提供 PB 级别的海量存储空间，并支持百万级别的文件读写操作，从而显著提升了整体系统的弹性和处理效能。

Swift 消息系统是在计算与存储分离上的一次尝试，其主要特点包括：

每个计算结点都是无状态的，即每个 worker 上除 log 记录，不存储任何消息系统相关数据。系统的状态数据存储在 Zookeeper 上，消息的内容则存储在分布式文件系统如 HDFS 上。
每个计算结点都是等价的，只要消息系统需要计算资源，就可以通过调度系统不停的申请并提升整个消息系统的服务能力。
Swift 自身的 client 与 server 的消息读写协议，能够保证消息高效可靠的传递。

三、Swift 系统结构

Swift 的系统结构中，主要分成 2 种 worker：Admin 和 Broker。Admin 和 Broker 的资源分配与启动都是基于调度系统。当前 Havenask 开源版本支持基于 Hippo 封装的 SSH 进行调度，这 2 种 worker 都会有很多个实例，Broker worker 都是等价的，Admin worker 则有一个 leader，其余的等价，这些 worker 一般在 Docker 容器中工作。

Admin 角色主要负责：

Topic 的增删改；
Topic 对应物理 Partition 与 Broker 调度；
Client 读写数据时物理 Partition 的定位；
资源的调整，如 Broker 个数的增减等。

Broker 角色主要负责：

Partition 相关的消息的读写；
Partition 相关数据的管理如过期数据的清理等。

四、Swift Topic 介绍

Swift 系统中的 Topic 与其它消息系统的类似，它是一堆相关消息的集合，通常由业务自定义。在 Swift 中，Topic 是由 65536 个逻辑分区组成，编号是[0 - 65535]。在 Swift 消息系统内部，Topic 是由 Partition 组成的，每个 Partition 负责一个 range 的逻辑 Partition 读写。

在用户层面，用户看不到 Swift 的物理 Partition，写消息时要么需要提供一个 hash 字段（由 Swift client 自动映射到相应的逻辑分区）要么提供一个 0-65535 的逻辑编号。Swift 根据 Topic 下每个 Partition 的服务 range，把消息写入相应 Partition 的 writer 中。Writer 可以通过同步与异步方式把消息 append 到对应的物理 Partition 中。

Topic 的物理 Partition 个数影响整个 Topic 的读写能力，通过逻辑 Partition 与物理 Partition 映射，当 Topic 的服务能力不足时，可以动态的扩展物理 Partition 来提升读写能力。另外，物理 Partition 是 Swift 的基本调度单元，Admin 会根据每个 Broker worker 负载，尽可能平衡地调度 Partition。

五、Swift 消息可靠传递机制

先前提到传统的消息系统为了保证消息的可靠性，在写消息时需要先落盘，以防机器挂掉时消息丢失。Swift 也提供类似的模式，但落盘的对象是分布式文件系统如 HDFS。这种模式下正常写落盘消息延时的毫秒级，当 HDFS 压力大时，会变成秒级，所以其性能不太稳定。

Swift 设计了一种 client 与 Broker 之间,Broker 与 HDFS 之间的消息写入与确认协议来保证消息高效可靠的写入与持久化，其机制类似 TCP 的滑动窗口协议。下图是消息异步安全发送的示意图。Broker 在分配到 Partition 进行服务时，会生成一个标记，其由 Partition 的版本号(V)，Broker 加载 Partition 时间戳(S)以及消息持久化的 checkpoint (C)组成。Client 在向 Admin 定位到 Partition 所在 Broker 的时候也会获取 Partition 的版本号(V)。版本号 V 主要在 Topic 属性发生变化时（例如 Partition 的个数等）会更新。时间戳在每次 Partition 发生重新加载或调度都会发生变化。

用户通过客户端写入一条消息，client 定位到写哪个物理 Partition，同时把消息写入到对应的 buffer 中。用户写消息时，还可以给每条消息设置一个递增编号，Swift client 会自动映射写消息进度与编号的关系。在异步模式中，client 会有专门的提交线程与 Broker 进行通信。

Client 第一次向 Partition 发送消息时，Broker 会验证 Partition 的版本 V0, 匹配后才会接受消息，同时会把三元组(V，S, C)返回。client 收到 accept 消息后，会更新已接受消息的光标和协议的三元组信息。

客户端可以持续地写入消息，同时 Broker 把 Partition 中的消息做异步持久化，当持久化成功时，会更新持久化信息(Ca)。持久化成功的消息在内存中不会马上删除，只有内存不足时才会被回收。

Client 的后台发送线程继续工作，发送消息 b，同时请求带上了(V0,S0)。

Broker 端验证(V0,S0)，收受消息 b，顺便把持久化信息也返回(V0,S0,Ca), client 接收到 accept 信息后，更新已发送的光标到 b，同时更新已接受的光标到 a。消息 a 已经持久化成功，在使用的内存将会被 writer 回收。Writer 更新 checkpoint (Ca)给用户层，表示消息 a 已经持久化。

同 3 一样，client 继续写消息 c，Broker 继续持久化消息 b。

此时 Partition 发生了调度（例如被分配到了其它机器），其 HDFS 上的文件消息马上可以读取到，但内存中的消息会被清空。此时 Partition 加载时间戳变成了 S1。Client 向 Admin 重新定位到 Partition 的服务 Broker 写入的消息 c 和(V0,S0)。

Broker 检查 client 发送的(V0,S0)与自身的(V0,S1)不相等，将拒绝此次消息的写入。主要基于消息在 Partition 内要求保序考虑。此时 client 还不知道 b 是否被序列化成功，Partition 重新被加载 b 是否被序列化成功的信息也会被丢弃(无状态)，所以它也不知道。因 Broker 返回(V0,S1,C0)，要求 client 重新发送未持久化的所有消息。

Client 重置已发送光标到 b 之前，更新 S1 并重新发送消息 b 和 c。

Broker 检验 client 的(V0,S1)并收受消息 b 和 c，这时消息 b 会被再次持久化到 HDFS 上。Client 重新更新已发送光标到 c。如果此后无新消息的写入，且 buffer 中的消息还有未被持久化的，client 会发起一次空写操作获取最新的持久化信息。

步骤 1-10 是异步消息写入的工作方式，用户层可以获取到当前持久化消息的 checkpoint，可以自己记录发送进度以便回滚。如果不方便记录发送进度，可以在写完一段数据后，调用 flush 方法强制把数据从 client 的 buffer 放到 Broker 的 buffer 中。此时消息虽然没有被持久化，但在 client 与 Partition 各存一份。所以只有在 Broker 与 client 同时挂掉才出现消息丢失，因此我们认为这种方法也是比较安全的。

Swift Partition 的写 buffer 缓存所有写入的消息，只有当空间不足时，消息内存空间才会被回收。对文件上的消息读取，也会以块 buffer 的方式做缓存。Partition 之间的 buffer 和文件 cache buffer 都是共享存储，由统一的回收模块管理。其保证冷门的 Partition 基本不消耗资源，热门的 Partition 可以利用非常的多资源。

正常情况下，Swift 的内存可以缓存 1-10 分钟的消息，所以消费消息时基本上从内存读取，读的性能也会很高效。在这个协议下，写 HDFS 发生偶尔抖动也不会影响消息的时效性，实际中 HDFS 在 10 分钟内的挂机也不影响消息的实时传递。

六、Swift Admin

相比较其他消息中间件，如 Kafka、Metaq，Swift 多了一个 Admin 的角色。Admin 的存在使得对集群的操作有统一的入口，系统的容错性也更为强大。

Swift Admin 的另一个重要功能就是进行 Partition 的动态调度，简单来说就是将所有的逻辑 Partition 向 Broker 上进行任务分配的过程。

七、Swift Broker

Broker 是 Swift 消息发送和读取的载体，由 Admin 通过调用 hippo 进行启动和调度。Broker 是单独的一个进程，单台物理机器可以运行多个 Broker。Broker 加载的最小单位是 Topic 的 Partition，一个进程内可以加载相同 Topic 的多个不同 Partition。Partition 由所在的分组名称、Topic 的名字以及列数所决定。

Broker 在接收消息时，默认先写到内存中，然后由后台线程进行消息的持久化，一般是写到 HDFS 上，这样可以保证 Broker 在迁移或者重启后仍然可以读到消息。读消息时也是先从内存消息中读取，读取不到再从 HDFS 中进行读取，以保证消息读取的效率。

Broker 与 Admin 之间通过 zookeeper 进行通信，Admin 会将需要 Broker 加载的 Partition 信息放到 zk 的指定路径的文件中，文件名即 Broker 机器地址，对应的 Broker 会监控此路径，当内容发生变化时进行任务的读取，即 Partition 的加载。同时 Broker 会实时向 Admin 汇报心跳，内容即加载 Partition 的信息，Admin 再根据 Broker 的心跳信息进行决策。

八、Swift Client

Swift Client 包括 Writer 和 Reader，分别介绍如下：

Swift Writer

用户在向 Swift 写消息的时候，消息从客户端的写出到最终 HDFS 的落盘，如下图所示，中间经历了两个 buffer，分别是客户端的 buffer 和 Broker 端的 buffer。而在确认结果时，也提供了 waitSent 和 waitFinish 两个接口来分别确认消息发送的位置。

Swift reader

Swift 在进行消息读取的时候，单 Partition 保证有序，多 Partition 顺序不做保证。Swift Reader 每次都会从单个 partition 对应的 SingleReader 中遍历找到拥有最小时间戳的消息，然后批量进行读取，存入 buffer 中等待用户消费，消费完成后，再次进行遍历获取最小时间戳的消息。所以 Swift 读消息时候不保证多 Partition 之间有序。

九、总结

Swift 作为 Havenask 消息系统的核心组件，通过计算与存储分离架构，提升系统的可扩展性和资源利用率。在 Swift 中，Topic 由多个逻辑分区组成，每个分区对应一个或多个物理 Partition，通过动态调度和映射，可以灵活调整 Topic 的读写能力。

Swift 采用了一种类似于 TCP 滑动窗口协议的机制来保证消息的高效可靠传递，即使在出现 Partition 迁移或 Broker 重启的情况下，也能确保消息不丢失且维持高时效性。此外，Swift 还引入了 Admin 角色，提供了一个集中式入口进行集群管理和操作，显著增强了系统的容错能力和运维效率。

同时，Swift 实现了对 Partition 的动态调度等多种优化，以适应现代分布式架构对于弹性、高并发和低延迟的需求。