阿里开源搜索引擎Havenask的消息系统

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

01 Havenask介绍

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

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

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

02 Swift简介

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

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

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

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

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

03 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相关数据的管理如过期数据的清理等。

04 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。

05 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分钟内的挂机也不影响消息的实时传递。

06 Swift Admin

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

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

07 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的心跳信息进行决策。

08 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之间有序。