队列托管 - 如果为mq消费者加一层代理会怎么样？

1. 引言

消息队列(Message Queue)，简称为MQ，是一种跨进程的通信机制。 其作为分布式系统中最常见也是最重要的组件之一，主要用于解决应用解耦、异步通信、流量削峰等问题。

通常，我们在业务中的使用方式如下(以Kafka为例)：

• • 生产端：基于kafka sdk创建一个producer，并调用SendMessage将消息推送到kafka的某个Topic
  • 消费端：基于kafka sdk创建一个comsumer，对于某个topic中的消息，执行拉取、处理、ack等操作

如上，kafka sdk耦合到了业务服务中，同时业务方需要完成生产者/消费者的创建，以及消息拉取/消息确认等业务不相关的逻辑[Q1]。

今天，这篇文章将介绍一种新的方案来优雅的解决上述问题 -- 队列托管(给生产/消费加一层代理服务)。其次，此方案还能提供更加强大的队列管理/消息控制能力，如：暂停消费/消息落盘&重放/并发控制/自动ack&处理成功ack等。

当然，上述问题[Q1]一定程度上也可以通过更加优秀的框架封装、增加防腐层设计等方式进行处理。

2. 队列托管概述

2.1. 队列消费的共性

在传统队列消费过程中，我们通常会进行以下步骤:

1. 创建队列客户端
2. 监听队列获取消息
3. 队列消息处理
4. offset提交

其中 1. 创建队列客户端 2. 监听队列 及 4. offset提交 都是一个比较通用的过程

2.2. 一种队列消费处理模型

在mq和事件处理服务之间，再加一个消费处理单元：

1. 将通用的消费逻辑封装成独立的消费单元
2. 业务方以http接口的方式提供事件处理逻辑，并由消费单元调用
3. 通过后台配置，提供灵活多样的消费逻辑定制
4. 根据后台配置，调度服务使用k8s api自动拉起消费进程

实现消费逻辑与业务处理的解耦，使业务更加专注于业务

2.3. 队列托管的优点

2.3.1. 业务开发

• 简化交互： 队列托管屏蔽了消息队列的复杂性，业务只需提供用于事件处理的http接口即可
• 多业务场景支持： 消费单元以配置的形式提供多种不同的消费模式，可满足不同业务场景的消费需求
• 配置化 & 自动化： 在后台更改配置，就可以一键完成消费单元的创建/销毁/消费逻辑变更等能力
• 提供丰富的辅助能力： 消息落盘，暂停消费，手动重放，并发处理等

2.3.2. 队列管理

• 提供管理队列的可视化界面
• 消费负载自动扩缩容(业务进程数量可摆脱原有patition数的限制)
• 消费启停/消费并发及速度，均可在后台调控
• 提供队列相关的监控面板
• 队列负责人机制：更好管理队列归属，消费告警中携带负责人字段，通知到人

2.3.3. 队列统一

• 业务侧无需关心底层使用的队列实例，选用不同的消费单元镜像即应用于不同的队列实例(如kafka，redis，RocketMQ)

3. 队列托管配置示例

3.1. 消费单元deployment

调度服务会根据当前配置自动化拉起消费单元进程

3.2. 消费单元处理配置展示

消费单元的定制化配置，以config map的方式挂载到消费单元中。

3.3. 常见消费场景下应用对比

	kafka	队列托管
kafka:自动提交偏移位托管：失败重试n次	消息最多拉取一次，如果处理失败则消息被丢弃	消息失败重试n次，如果依然处理失败，则写入数据库，支持手动重放
kafka: 手动提交偏移位托管：失败一直重试	消息一直重试，直到处理成功	消息一直重试，直到处理成功。如果出现无法消费的异常消息：方案一：切到最小损失放过这条消息方案二：使用错误屏蔽放过这条消息
并发数量	kafka的消费者数量受限于patition数	消费单元的数量受限于patition，但是事件处理服务的数量不受限
其他功能	无	失败重试消息重放消费暂停消息落盘...

备注：队列托管是通过增加一层消费代理的方式实现的，也就是说多了一层数据转换&请求转发的损耗，不太适用于日志采集类场景。

4. 队列托管架构实现

4.1. 主体模块设计

1. 管理后台：用于管理队列、消息、消费者等配置
2. 调度服务：调度消费单元，用于管理k8s中消费单元相关的deployment和config_map
3. 消费单元：由消费配置创建的消费单元进程，用于拉取消息并调用http接口触发业务事件处理
4. 生产托管：此节不介绍

4.2. 系统架构设计

controller从配置管理服务拉取消费配置，并更新k8s中的deployment和config，由k8s拉起消费单元容器。

4.3. 核心消费逻辑演示

k8s拉起消费单元容器后，消费单元进程会读取本地的config map配置，完成如下组件的组装：

• mq消费对象
• 消费处理器(并行 or 串行，自动ack or 成功ack，重试策略等)
• 事件触发器(http trigger)
• 消息落盘器(msg store)

5. 稳定性介绍

服务稳定性：

• k8s部署 - 消费单元崩溃可自动恢复
• 消费单元数/单节点消费并发/消费启停，支持后台界面配置，即配即用
• 失败消息落盘后，支持手动重试
• 最后兜底：消费失败的消息内容会打印一份到日志中

• 监控面板：由消费单元打点并上报消费情况，用于绘制消费监控面板

告警：

• 消费失败告警：只要有一条消息消费失败则告警(支持推送到负责人)
• 服务异常告警：总体日志错误数超过5条告警

6. 已支持功能

1. 消费暂停：管理后台可一键暂停消费
2. 消息重试：对于自动ack类型的消息，支持失败重试n次
3. 消费串行&并行：同一时间内，1个patition可以设置n个消息在并发处理
4. 错误屏蔽：有些情况下，由于业务设计问题，即使返回了错误，也应正常消费这条消息

• • 如：积分为0的变更事件，业务处理时返回了错误：change=0无需变更

1. 消息落盘
2. 手动重放：搭配消息落盘能力，实现任意消息的重放能力
3. 延迟消息：用于固定时长的延迟消息的场景

7. 结尾

在这篇分享中，我们探讨了消息队列在分布式系统中的重要性以及常见的使用方式。通过引入队列托管这一新的解决方案，我们成功地解耦了业务服务和消息队列的关系，提高了系统的灵活性和可维护性。

队列托管不仅简化了业务开发的流程，还提供了丰富的队列管理功能，使得队列的配置和调整更加灵活方便。通过对比常见的消费场景，我们发现队列托管在处理失败消息、管理并发数量等方面具有明显的优势。同时，我们也介绍了队列托管的架构设计和稳定性保障措施，确保了系统的稳定运行和高可用性。

综上所述，队列托管为我们构建高效可靠的分布式系统提供了一种全新的选择，在实际应用中也取得了显著的效果。

没有什么是加一层封装解决不了的，如果有，那就再加一层～