一篇文了解分布式队列编程：从模型、实战到优化

欢迎大家来到IT世界,在知识的湖畔探索吧!

“

本文由美团点评技术团队出品，一篇文助你掌握分布式队列编程的要义。从模型到实战再到优化，基本涵盖你可能踩的坑与其解决办法。

• 使用分布式队列的时候，没有意识到它是队列。
• 有具体需求的时候，忘记了分布式队列的存在。

文章首先从最基础的需求出发，详细剖析分布式队列编程模型的需求来源、定义、结构以及其变化多样性。通过这一部分的讲解，作者期望能在两方面帮助读者：一方面，提供一个系统性的思考方法，使读者能够将具体需求关联到分布式队列编程模型，具备进行分布式队列架构的能力；另一方面，通过全方位的讲解，让读者能够快速识别工作中碰到的各种分布式队列编程模型。

老司机简介

刘丁，新美大广告平台CRM系统技术负责人，曾就职于Amazon、Tripadvisor。2014年加入美团，先后负责美团推荐系统、智能筛选系统架构，作为技术负责人主导了美团广告系统的开发和上线。目前致力于推进新美大广告运营的标准化、自动化和智能化。

新美大广告平台是美团、大众点评双平台的营销推广平台，帮助商户推广店铺品牌及提升客流量。

1模型篇

何时选择分布式队列

When：同步VS异步

• 发出去的消息是否需要确认，如果不需要确认，更像是异步通信，这种通信有时候也称为单向通信（One-Way Communication）。

• 如果需要确认，可以根据需要确认的时间长短进行判断。时间长的更像是异步通信，时间短的更像是同步通信。当然时间长短的概念是纯粹的主观概念，不是客观标准。
• 发出去的消息是否阻塞下一个指令的执行，如果阻塞，更像是同步，否则，更像是异步。

无论如何，工程师们不能生活在混沌之中，不做决定往往是最坏的决定。当分析一个通信需求或者进行通信构架的时候，工程师们被迫作出“同步”还是“异步”的决定。当决策的结论是“异步通信”的时候，分布式队列编程模型就是一个备选项。

Who：发送者接收者解耦

• 无论是发送方还是接收方，只需要跟消息中间件通信，接口统一。统一意味着降低开发成本。

• 在不影响性能的前提下，同一套消息中间件部署，可以被不同业务共享。共享意味着降低运维成本。
• 发送方或者接收方单方面的部署拓扑的变化不影响对应的另一方。解藕意味着灵活和可扩展。

Where：消息暂存机制

How：如何传递

• 可用性，如何保障通信的高可用。
• 可靠性，如何保证消息被可靠地传递。
• 持久化，如何保证消息不会丢失。
• 吞吐量和响应时间。
• 跨平台兼容性。
• 除非工程师对造轮子有足够的兴趣，并且有充足的时间，采用一个满足各项指标的分布式队列编程模型就是一个简单的选择。

分布式队列编程定义

需要重点明确的概念是分布式队列，它是提供以下功能的应用程序或服务：

1. 接收“发送者”产生的消息实体；
2. 传输、暂存该实体；
3. 为“接收者”提供读取该消息实体的功能。

特定的场景下，它当然可以是Kafka、RabbitMQ等消息中间件。但它的展现形式并不限于此，例如：

• 队列可以是一张数据库的表，发送者将消息写入表，接收者从数据表里读消息。

• 如果一个程序把数据写入Redis等内存Cache里面，另一个程序从Cache里面读取，缓存在这里就是一种分布式队列。

• 流式编程里面的的数据流传输也是一种队列。
• 典型的MVC（Model–view–controller）设计模式里面，如果Model的变化需要导致View的变化，也可以通过队列进行传输。这里的分布式队列可以是数据库，也可以是某台服务器上的一块内存。

抽象模型

点对点模型（Point-to-point）

生产者消费者模型（Producer–consumer）

发布订阅模型（PubSub）

MVC模型

编程模型

分布式队列模型编程和异步编程

分布式队列模式编程和流式编程

首先，本文的队列编程模式不依赖于任何框架，而流式编程是在具体的流式框架内的编程。

其次，分布式队列编程模型是一个需求解决方案，关注如何根据实际需求进行分布式队列编程建模。流式框架里的数据流一般都通过队列传递，不过，流式编程的关注点比较聚焦，它关注如何从流式框架里获取消息流，进行map、reduce、 join等转型（Transformation）操作、生成新的数据流，最终进行汇总、统计。

2实战篇

这里所有的项目都是作者在新美大工作的真实案例。实战篇的关注点是训练建模思路，所以这些例子都按照挑战、构思、架构三个步骤进行讲解。受限于保密性要求，有些细节并未给出，但这些细节并不影响讲解的完整性。

另一方面，特别具体的需求容易让人费解，为了使讲解更加顺畅，作者也会采用一些更通俗易懂的例子。通过本篇的讲解，希望和读者一起去实践“如何从需求出发去构架分布式队列编程模型”。

需要声明的是，这里的解决方案并不是所处场景的最优方案。但是，任何一个稍微复杂的问题，都没有最优解决方案，更谈不上唯一的解决方案。实际上，工程师每天所追寻的只是在满足一定约束条件下的可行方案。当然不同的约束会导致不同的方案，约束的松弛度决定了工程师的可选方案的宽广度。

信息采集处理

• 采集者和处理者解耦，采集发生在客户端，而计费发生在服务端。

• 计费与钱息息相关。

• 重复计费意味着灾难。

• 计费是动态实时行为，需要接受预算约束，如果消耗超过预算，则广告投放需要停止。
• 用户的浏览和点击量非常大。

挑战

• 高吞吐量－－广告的浏览和点击量非常巨大，我们需要设计一个高吞吐量的采集架构。

• 高可用性－－计费信息的丢失意味着直接的金钱损失。任何处理服务器的崩溃不应该导致系统不可用。

• 高一致性要求－－计费是一个实时动态处理过程，但要受到预算的约束。收集到的浏览和点击行为如果不能快速处理，可能会导致预算花超，或者点击率预估不准确。所以采集到的信息应该在最短的时间内传输到计费中心进行计费。

• 完整性约束－－这包括反作弊规则，单个用户行为不能重复计费等。这要求计费是一个集中行为而非分布式行为。
• 持久化要求－－计费信息需要持久化，避免因为机器崩溃而导致收集到的数据产生丢失。

构思

架构

• 用户点击浏览收集服务（Click/View Collector）作为生产者部署在多个机房里，以提高收集服务可用性。

• 每个机房里采集到的数据通过消息队列中间件发送到核心机房IDC_Master。

• Billing服务作为消费者部署在核心机房集中计费。

• 提高可扩展性，如果一个Billing部署实例在性能上无法满足要求，可以对采集的数据进行主题分区（Topic Partition）计费，即采用发布订阅模式以提高可扩展性（Scalability）。

• 全局排重和反作弊。采用集中计费架构解决了点击浏览排重的问题，另一方面，这也给反作弊提供了全局信息。
• 提高计费系统的可用性。采用下文单例服务优化策略，在保障计费系统集中性的同时，提高计费系统可用性。

分布式缓存更新（Distributed Cache Replacement）

• 接收到请求后，先读取缓存，如果命中则返回结果。

• 如果缓存不命中，读取DB或其它持久层服务，更新缓存并返回结果。

挑战

• 数据一致性低。分布式缓存中键值数量巨大，从而导致LRU或者LFU算法更新周期很长。在分布式缓存中，拿LRU算法举例，其典型做法是为每个Key值设置一个生存时间（TTL），生存时间到期后将该键值从缓存中驱逐除去。

  考虑到分布式缓存中庞大的键值数量，生存时间往往会设置的比较长，这就导致缓存和持久层数据不一致时间很长。如果生存时间设置过短，大量请求无法命中缓存被迫读取持久层，系统响应时间会急剧恶化。

• 新数据不可用。在很多场景下，由于分布式缓存和持久层的访问性能相差太大，在缓存不命中的情况下，一些应用层服务不会尝试读取持久层，而直接返回空结果。漫长的缓存更新周期意味着新数据的可用性就被牺牲了。从统计的角度来讲，新键值需要等待半个更新周期才会可用。

• 提高可扩展性，如果一个Cache Updater在性能上无法满足要求，可以对键值进行主题分区（Topic Partition）进行并行缓存更新，即采用发布订阅模式以提高可扩展性（Scalability）。
• 更新频率控制。缓存更新都集中处理，对于发布订阅模式，同一类主题（Topic）的键值集中处理。Cache Updater可以控制对同一键值的在短期内的更新频率（参见下文排重优化）。

后台任务处理

挑战

• 数据一致性问题。以火车票预订为例，用户筛选火车票和最终购买之间往往有一定的时延，意味着两个操作之间数据是不一致的。在筛选阶段，工程师们需决定是否进行车票锁定，如果不锁定，则无法保证出票成功。反之，如果在筛选地时候锁定车票，则会大大降低系统效率和出票吞吐量。

• 约束问题。工单创建需要满足很多约束，主要包含两种类型：动态约束，与操作者的操作行为有关，例如购买几张火车票的决定往往发生在筛选最后阶段。隐性约束，这种约束很难通过界面进行展示，例如一个用户购买了5张火车票，这些票应该是在同一个车厢的临近位置。

• 优化问题。工单创建往往是约束下的优化，这是典型的统筹优化问题，而统筹优化往往需要比较长的时间。
• 响应时间问题。对于多任务工单，一个请求意味着多个任务产生。这些任务的创建往往需要遵循事务性原则，即All or Nothing。在数据层面，这意味着工单之间需要满足串行化需求（Serializability）。大数据量的串行化往往意味着锁冲突延迟甚至失败。无论是延迟机制所导致的长时延，还是高创建失败率，都会大大伤害用户体验。

• 用户首选进行规则创建，这个过程主要是一些搜索筛选操作。

• 用户点击工单创建，TicketRule Generator将把所有的筛选性组装成规则消息并发送到队列里面去。

• Ticket Generator作为一个消费者，实时从队列中读取工单创建请求，开始真正创建工单。

• 数据锁定推迟到工单创建阶段，可以减少数据锁定范围，最大程度的降低工单创建对其他在线操作的影响范围。
• 如果需要进行统筹优化，可以将Ticket Generator以单例模式进行部署（参见单例服务优化）。这样，Ticket Generator可以读取一段时间内的工单请求，进行全局优化。

  例如，在我们的项目中，在某种条件下，运营人员需要满足分级公平原则，即相同级别的运营人员的工单数量应该接近，不同级别的运营人员工单数量应该有所区分。如果不集中进行统筹优化，实现这种优化规则将会很困难。

• 保障了约束完整性。例如，在我们的场景里面，每个运营人员每天能够处理的工单是有数量限制的，如果采用并行处理的方式，这种完整性约束将会很难实施。

3优化篇

接下来重点阐述工程师运用分布式队列编程构架的时候，在生产者、分布式队列以及消费者这三个环节的注意点以及优化建议。

生产者优化

缓存优化

处于“处理－转发”模式下运行的生产者往往被设计成请求驱动型的服务，即每个请求都会触发一个处理线程，线程处理完后将结果写入分布式队列。如果由于某种原因队列服务不可用，或者性能恶化，随着新请求的到来，生产者的处理线程就会产生堆积。这可能会导致如下两个问题：

• 系统可用性降低。由于每个线程都需要一定的内存开销，线程过多会使系统内存耗尽，甚至可能产生雪崩效应导致最终完全不可用。
• 信息丢失。为了避免系统崩溃，工程师可能会给请求驱动型服务设置一个处理线程池，设置最大处理线程数量。这是一种典型的降级策略，目的是为了系统崩溃。

  但是，后续的请求会因为没有处理线程而被迫阻塞，最终可能产生信息丢失。例如：对于广告计费采集，如果采集系统因为线程耗尽而不接收客户端的计费行为，这些计费行为就会丢失。

缓解这类问题的思路来自于CAP理论，即通过降低一致性来提高可用性。生产者接收线程在收到请求之后第一时间不去处理，直接将请求缓存在内存中（牺牲一致性），而在后台启动多个处理线程从缓存中读取请求、进行处理并写入分布式队列。

与线程所占用的内存开销相比，大部分的请求所占内存几乎可以忽略。通过在接收请求和处理请求之间增加一层内存缓存，可以大大提高系统的处理吞吐量和可扩展性。这个方案本质上是一个内存生产者消费者模型。

批量写入优化

如果生产者的请求过大，写分布式队列可能成为性能瓶颈，有如下几个因素：

• 队列自身性能不高。

• 分布式队列编程模型往往被应用在跨机房的系统里面，跨机房的网络开销往往容易成为系统瓶颈。
• 消息确认机制往往会大大降低队列的吞吐量以及响应时间。

如果在处理请求和写队列之间添加一层缓存，消息写入程序批量将消息写入队列，可以大大提高系统的吞吐量。原因如下：

• 批量写队列可以大大减少生产者和分布式队列的交互次数和消息传输量。特别是对于高吞吐小载荷的消息实体，批量写可以显著降低网络传输量。
• 对于需要确认机制的消息，确认机制往往会大大降低队列的吞吐量以及响应时间，某些高敏感的消息需要多个消息中间件代理同时确认，这近一步恶化性能。在生产者的应用层将多条消息批量组合成一个消息体，消息中间件就只需要对批量消息进行一次确认，这可能会数量级的提高消息传输性能。

持久化优化

通过添加缓存，消费者服务的吞吐量和可用性都得到了提升。但缓存引入了一个新问题——内存数据丢失。对于敏感数据，工程师需要考虑如下两个潜在问题：

• 如果内存中存在未处理完的请求，而某些原因导致生产者服务宕机，内存数据就会丢失而可能无法恢复。
• 如果分布式队列长时间不可用，随着请求数量的不断增加，最终系统内存可能会耗尽而崩溃，内存的消息也可能丢失。

所以缓存中的数据需要定期被持久化到磁盘等持久层设备中，典型的持久化触发策略主要有两种：

• 定期触发，即每隔一段时间进行一次持久化。

• 定量触发，即每当缓存中的请求数量达到一定阈值后进行持久化。
• 是否需要持久化优化，以及持久化策略应该由请求数据的敏感度、请求量、持久化性能等因素共同决定。

中间件选型

中间件的功能

• 消息接收

• 消息分发

• 消息存储
• 消息读取

概念模型

• 发送者和接收者客户端（Sender/Receiver Client），在具体实施过程中，它们一般以库的形式嵌入到应用程序代码中。

• 代理服务器（Broker Server），它们是与客户端代码直接交互的服务端代码。

• 消息交换机（Exchanger），接收到的消息一般需要通过消息交换机（Exchanger）分发到具体的消息队列中。

• 消息队列，一般是一块内存数据结构或持久化数据。

• 概念模型如下图：

• 为了提高分发性能，很多消息中间件把消息代理服务器的拓扑图发送到发送者和接收者客户端（Sender/Receiver Client），如此一来，发送源可以直接进行消息分发。

选型标准

性能

对于同一种中间件，不同的配置方式也会影响性能。主要有如下几方面的配置：

• 是否需要确认机制，即写入队列后，或从队列读取后，是否需要进行确认。确认机制对响应时间的影响往往很大。

• 能否批处理，即消息能否批量读取或者写入。批量操作可以大大减少应用程序与消息中间件的交互次数和消息传递量，大大提高吞吐量。

• 能否进行分区（Partition）。将某一主题消息队列进行分区，同一主题消息可以有多台机器并行处理。这不仅仅能影响消息中间件的吞吐量，还决定着消息中间件是否具备良好的可伸缩性（Scalability）。
• 是否需要进行持久化。将消息进行持久化往往会同时影响吞吐量和响应时间。

可靠性

高可用性的消息中间件应该具备如下特征：

• 消息中间件代理服务器（Broker）具有主从备份。即当一台代理服务宕机之后，备用服务器能接管相关的服务。

• 消息中间件中缓存的消息是否有备份、并持久化。
• 根据CAP理论，高可用、高一致性以及网络分裂不可兼得。根据作者的观察，大部分的消息中间件在面临网络分裂的情况下下，都很难保证数据的一致性以及可用性。 很多消息中间件都会提供一些可配置策略，让使用者在可用性和一致性之间做权衡。

高可靠的消息中间件应该确保从发送者接收到的消息不会丢失。中间件代理服务器的宕机并不是小概率事件，所以保存在内存中的消息很容易发生丢失。大部分的消息中间件都依赖于消息的持久化去降低消息丢失损失，即将接收到的消息写入磁盘。即使提供持久化，仍有两个问题需要考虑：

• 磁盘损坏问题。长时间来看，磁盘出问题的概率仍然存在。

• 性能问题。与操作内存相比，磁盘I/O的操作性能要慢几个数量级。频繁持久化不仅会增加响应时间，也会降低吞吐量。
• 解决这两个问题的一个解决方案就是：多机确认，定期持久化。即消息被缓存在多台机器的内存中，只有每台机器都确认收到消息，才跟发送者确认（很多消息中间件都会提供相应的配置选项，让用户设置最少需要多少台机器接收到消息）。由于多台独立机器同时出故障的概率遵循乘法法则，指数级降低，这会大大提高消息中间件的可靠性。

确认机制本质上是通讯的握手机制（Handshaking）。如果没有该机制，消息在传输过程中丢失将不会被发现。高敏感的消息要求选取具备确认机制的消息中间件。当然如果没有接收到消息中间件确认完成的指令，应用程序需要决定如何处理。典型的做法有两个：

• 多次重试。
• 暂存到本地磁盘或其它持久化媒介。

客户端接口所支持语言

投递策略（Delivery policies）

在实际应用中，只考虑消息中间件的投递策略并不能保证业务的投递策略，因为接收者在确认收到消息和处理完消息并持久化之间存在一个时间窗口。例如，即使消息中间件保证仅有一次（Exactly Once），如果接收者先确认消息，在持久化之前宕机，则该消息并未被处理。

从应用的角度，这就是最多一次（At most Once）。反之，接收者先处理消息并完成持久化，但在确认之前宕机，消息就要被再次发送，这就是最少一次（At least Once）。 如果消息投递策略非常重要，应用程序自身也需要仔细设计。

消费者优化

挑战

有序性

采用多消费者架构，这两条记录被两个消费者（Consumer1和Consumer2）处理后更新到数据库里面。Consumer1虽然先读取ri1但是却后写入数据库，这就导致，新的状态被老的状态覆盖，所以多消费者不保证数据的有序性。

串行化

频次控制

• 费用问题。如果每次消费所引起的操作都需要收费，而同一个请求消息在队列中保存多份，不进行频次控制，就会导致无谓的浪费。
• 性能问题。每次消费可能会引起对其他服务的调用，被调用服务希望对调用量有所控制，对同一个请求消息的多次访问就需要有所控制。

完整性和一致性

单例服务优化

领导人选举架构

Let ELECTION be a path of choice of the application. To volunteer to be a leader:

Upon receiving a notification of znode deletion:

领导人交接架构

类似的，为了解决领导人交接问题，所有的消费者从代码实现的角度都需要实现类似ILeaderCareer接口。这个接口包含三个方发inaugurate，handOver和execute。某个部署实例（Learner）在得知自己承担领导人角色后，需要调用inaugurate方法，进行加冕。主要的消费逻辑通过不停的执行execute实现，当确认自己不再承担领导人之后，执行handOver进行交接。

如果承担领导人角色的消费者，在执行execute阶段得知自己将要下台，根据消息处理的原子性，该领导人可以决定是否提前终止操作。如果整个消息处理是一个原子性事务，直接终止该操作可以快速实现领导人换届。否则，前任领导必须完成当前消息处理后，才进入交接阶段。这意味着新的领导人，在inaugurate阶段需要进行一定时间的等待。

排重优化

模型

重复度模型

是否需要进行排重优化取决于队列中请求的重复度。由于不同请求之间并不存在重复的问题，不失一般性，这里的模型只考了单个请求的重复度，重复度分为三个类：无重复、稀疏重复、高重复。

• 无重复：在整个请求过程，没有任何一个请求出现一次以上。

• 稀疏重复：主要的请求最小重复长度大于消费队列长度。
• 高重复：大量请求最小重复长度小于消费队列长度。

对于不同的重复度，会有不同的消费模型。

无重复消费模型

稀疏重复消费模型

高重复消费模型

排重状态机

由于状态机的处理单元是请求，所以需要针对每一个请求建立一个排重状态机。基于以上要求，我们设计的排重状态机包含4个状态Init，Process，Block，Decline。各个状态之间转化过程如下图：

1. 状态机创建时处于Init状态。
2. 对Init状态进行Enqueue操作，即接收一个请求，开始处理（称为头部请求），状态机进入Process状态。
3. 状态机处于Process状态，表明当前有消费者正在处理头部请求。此时，如果进行Dequeue操作，即头部请求处理完成，返回Init状态。如果进行Enqueue操作，即另一个消费者准备处理同一个请求，状态机进入Block状态（该请求称为尾部请求）。
4. 状态机处于Block状态，表明头部请求正在处理，尾部请求处于阻塞状态。此时，进行Dequeue操作，即头部请求处理完成，返回Process状态，并且尾部请求变成头部请求，原尾部请求消费者结束阻塞状态，开始处理。进行Enqueue操作，表明一个新的消费者准备处理同一个请求，状态机进入Decline状态。
5. 状态机进入Decline状态，根据等待唯一性目标，处理最新请求的消费者将被抛弃该消息，状态机自动转换回Block状态。

构思

一致性问题

完整性问题

完整性要求保障状态机Init，Process，Block，Decline四种状态正确、状态之间的转换也正确。由于状态机的操作非常轻量级，兼顾完整性和降低代码复杂度，我们对状态机的所有方法进行加锁。

请求缓存驱逐问题（Cache Eviction）

• 当状态机返回Init状态时，清除出队列。
• 启动一个后台线程，定时扫描状态机队列，采用LRU等标准缓存清除机制。

标识问题

实施

状态机实施（TrafficAutomate）

enQueue操作

deQueue操作

对于deQueue操作，首先将尾部请求赋值给头部请求，并将尾部请求置为无效。deQueue代码如下：

状态机队列实施(QueueCoordinator)

接口定义

状态机队列集中管理所有请求的排重状态机，所以其操作和单个状态机一样，即enQueue和deQueuqe接口。这两个接口的实现需要识别特定请求的状态机，所以它们的入参应该是请求。为了兼容不同类型的请求消息，我们采用了Java泛型编程。接口定义如下：

enQueue操作

enQueue操作过程如下：

首先，根据传入的请求key值，获取状态机， 如果不存在则创建一个新的状态机，并保存在ConcurrentHashMap中。

接下来，获取线程id作为该消费者的唯一标识，并对对应状态机进行enQueue操作。

在某些情况下，头部请求线程可能由于异常，未能对状态机进行deQueue操作（作为组件提供方，不能假定所有的规范被使用方实施）。为了避免处于阻塞状态的消费者无期限地等待，建议对状态机设置安全超时时限。超过了一定时间后，状态机强制清空头部请求，返回到业务层，业务层开始处理该请求。

代码如下：

deQueue操作

deQueue操作首先从ConcurrentHashMap获取改请求所对应的状态机，接着获取该线程的线程id，对状态机进行deQueue操作。

enQueue代码如下：

源代码

完整源代码可以在QueueCoordinator获取。链接：

https://github.com/dinglau2008/QueueCoordinator/tree/master/src

本文由 “美团点评技术团队”官方微信公众号，ID：meituantech 授权转载

▽

免费在线课堂报名！本期InfoQ在线课堂将邀请前EMC大中国区资深技术顾问、现任AWS中国资深技术讲师张波，围绕云计算自动化部署的实现以及AWS CloudFormation实践应用案例进行探讨与分享，对比传统与云端两种自动化部署及管理方式的不同和特点，深入讲解如何使用AWS CloudFormation实现基础设施的代码化，8月16日周二晚8点正式进行直播。

戳阅读原文，免费报名！

今晚八点半，直播等你来！

欢迎大家来到IT世界,在知识的湖畔探索吧!