背景
为什么使用消息队列
在kafka生产中，基本都有一个leader和多个follwer。follwer会去同步leader的信息。 因此，为了避免生产者丢数据，做如下两点配置：
通过副本和 ISR 保证数据高可用；
针对消息队列丢数据的情况，无外乎就是，数据还没同步，leader就挂了，这时zookpeer会将其他的follwer切换为leader,那数据就丢失了。
针对这种情况，应该做两个配置。
这两个配置加上上面生产者的配置联合起来用，基本可确保kafka不丢数据。
这种情况一般是自动提交了offset，然后你处理程序过程中挂了。kafka以为你处理好了。
offset：指的是kafka的topic中的每个消费组消费的下标。
简单的来说就是一条消息对应一个offset下标，每次消费数据的时候如果提交offset，那么下次消费就会从提交的offset加一那里开始消费。
比如一个topic中有100条数据，我消费了50条并且提交了，那么此时的kafka服务端记录提交的offset就是49(offset从0开始)，那么下次消费的时候offset就从50开始消费。
解决方案也很简单，改成手动提交offset即可。
| 阶段 | 目标 | 关键措施 |
|---|---|---|
| 紧急止损 | 快速恢复，避免事态恶化 | 定位瓶颈：监控生产/消费qps，单条耗时，分区进度 |
| 根源优化 | 提升系统处理能力 | |
| 长期预防 | 杜绝问题反复发生 |
首先确认积压是否真实存在，并判断影响范围。
重点关注以下指标：
| 指标 | 说明 |
|---|---|
| 消息堆积数量 | 当前未消费消息总量 |
| 最老消息年龄 | 队列中最早未消费消息等待了多久 |
| 生产速率 | 每秒新增消息数量 |
| 消费速率 | 每秒成功消费消息数量 |
| 消费失败率 | 消费异常、重试、超时比例 |
| 消费延迟 | 消息从生产到消费完成的时间 |
| 消费者实例数 | 当前正常运行的消费者数量 |
| 分区积压分布 | 是否只有部分分区或队列积压 |
| 下游依赖状态 | 数据库、缓存、接口等是否异常 |
需要优先回答以下问题：
当消费速度明显低于生产速度时，第一目标不是盲目扩容，而是先阻止积压继续扩大。
可采取以下措施：
根据业务优先级限制消息生产速度：
原则：
消费速率无法恢复前，不要让生产速率无限增长。
将消息划分为不同优先级：
优先保障核心消息，必要时暂停或丢弃低优先级消息。
如果消息中间件支持优先级队列，可直接使用优先级机制；如果不支持，可拆分为不同 Topic 或 Queue。
消费链路中可能包含多个非核心操作，例如：
故障期间可临时关闭非必要步骤，只保留核心处理逻辑。
例如：
text正常流程： 消费消息 → 写数据库 → 更新缓存 → 调用第三方接口 → 记录统计 → 发送通知 应急流程： 消费消息 → 写数据库 → 结束 后续通过补偿任务完成缓存、通知和统计处理。
优先排查消费者是否存在以下问题：
紧急处理方式包括：
注意：
重启只能解决临时性故障。如果根因是代码缺陷、容量不足或下游瓶颈，单纯重启会再次积压。
扩容是处理积压最直接的方法之一，但扩容前必须确认消费能力是否能随实例数量线性增长。
扩容前检查：
例如，Kafka 一个分区在同一消费组内通常只能被一个消费者实例消费。
如果：
textTopic 分区数 = 8 消费者实例数 = 8
继续扩容到 16 个实例，通常不会带来明显提升，因为多余实例无法获得分区。
正确做法可能是：
text增加分区数 → 扩容消费者 → 验证分区数据是否均衡
但增加分区可能影响消息顺序，需要提前评估。
如果单条消息处理开销较大，可临时切换为批量模式。
例如：
text单条消费： 读取 1 条消息 → 写数据库 1 次 → 提交消费位点 1 次 批量消费： 读取 500 条消息 → 批量写数据库 → 批量提交消费位点
批量处理可减少：
需要注意：
某些积压由“毒消息”引起，即特定消息持续处理失败，导致消费者不断重试。
常见异常消息包括：
处理方式：
示例：
text正常队列 → 消费失败 → 重试队列 → 达到最大重试次数 → 死信队列 → 人工修复或补偿任务处理
避免以下模式：
text失败 → 立即重试 → 再次失败 → 无限重试
无限重试会放大故障，并阻塞正常消息。
当积压量非常大，而原消费链路短时间内无法消化时，可建立临时消费集群。
临时集群可采用简化逻辑：
text消息队列 → 临时消费者集群 → 写入临时存储 → 后台慢速处理
例如先将消息快速落入：
这样可以快速释放原始消息队列容量，再通过离线任务逐步处理。
适用场景：
积压恢复过程中，消费者可能向下游集中发送大量请求。
需要设置恢复速率上限：
text恢复消费速率 ≤ 下游系统可承受速率
可以采用：
推荐采用“双通道模式”：
text实时通道：优先处理新消息 历史通道：限速处理积压消息
这样可以避免新消息继续等待在旧消息之后。
可使用以下公式估算：
text净消费速率 = 消费速率 - 生产速率 预计清理时间 = 当前积压消息数 ÷ 净消费速率
示例：
text当前积压：3,600,000 条 生产速率：1,000 条/秒 消费速率：2,000 条/秒 净消费速率： 2,000 - 1,000 = 1,000 条/秒 预计清理时间： 3,600,000 ÷ 1,000 = 3,600 秒 约 1 小时
如果消费速率小于或等于生产速率：
text消费速率 ≤ 生产速率
则积压永远无法清理完，必须限流、优化或扩容。
markdown- [ ] 确认积压 Topic、Queue、消费组和分区范围
- [ ] 确认当前生产速率和消费速率
- [ ] 确认最老消息等待时间
- [ ] 检查消费者实例是否正常
- [ ] 检查最近是否发生发布或配置变更
- [ ] 检查数据库、缓存和外部接口状态
- [ ] 暂停非核心生产任务
- [ ] 对生产端进行限流
- [ ] 回滚异常版本
- [ ] 隔离毒消息
- [ ] 开启死信队列
- [ ] 临时扩容消费者
- [ ] 验证分区数是否支持扩容
- [ ] 开启批量消费和批量写入
- [ ] 对历史积压消息限速回放
- [ ] 防止下游系统被恢复流量打垮
- [ ] 持续计算预计积压清理时间
积压问题通常可归为两类。
可能原因：
可能原因：
需要结合历史基线进行判断：
text积压变化 + 生产速率变化 + 消费速率变化 + 单条处理耗时变化 + 错误率变化
不要只统计“消费接口总耗时”，应将消费过程拆解。
例如：
text总耗时 = 拉取消息耗时 + 反序列化耗时 + 参数校验耗时 + 数据库查询耗时 + 数据库写入耗时 + 缓存访问耗时 + 外部接口调用耗时 + 消费结果提交耗时
建议记录以下指标：
| 环节 | 建议指标 |
|---|---|
| 消息拉取 | 拉取条数、拉取耗时 |
| 反序列化 | 失败率、平均耗时 |
| 业务处理 | 平均值、P95、P99 |
| 数据库查询 | QPS、慢查询、连接池等待时间 |
| 数据库写入 | 单条耗时、批量耗时、锁等待 |
| 外部接口 | 成功率、超时率、P99 |
| 位点提交 | 成功率、提交耗时 |
| 重试 | 重试次数、重试消息比例 |
通过链路追踪定位真正瓶颈，避免无效扩容。
数据库通常是消息消费链路中最常见的瓶颈。
低效模式：
text消费 1000 条消息 → 查询数据库 1000 次 → 更新数据库 1000 次
优化模式：
text消费 1000 条消息 → 批量查询 → 内存中完成匹配 → 批量新增或更新
对于稳定性较高的数据，可使用：
重点检查：
避免在长事务中执行：
推荐：
text先完成数据准备 → 开启事务 → 执行必要数据库操作 → 快速提交事务 → 执行非事务操作
批量写入需要控制批次大小。
例如：
text每批 200～1000 条
实际大小应根据以下因素压测确定：
消费链路同步调用外部接口时，很容易因外部系统抖动产生积压。
优化方式：
必须设置：
禁止无限等待。
当外部依赖异常时：
text调用失败率达到阈值 → 熔断 → 停止持续调用 → 消息进入补偿队列 → 外部系统恢复后重新处理
原流程：
text消费消息 → 调用外部接口 → 等待响应 → 提交消费结果
优化后：
text消费消息 → 写入本地任务表或二级队列 → 提交消费结果 异步任务 → 调用外部接口 → 失败重试
推荐使用指数退避：
text第 1 次重试：1 秒后 第 2 次重试：5 秒后 第 3 次重试：30 秒后 第 4 次重试：5 分钟后
同时设置：
低效模式：
text拉取一条 → 处理一条 → 提交一条
可优化为：
text批量拉取 → 线程池并发处理 → 批量提交
CPU 密集型任务包括：
线程数不宜远高于 CPU 核数。
I/O 密集型任务包括：
可适当增加并发，但需要受下游容量约束。
不要让轻量消息与重型消息共用同一个消费通道。
例如拆分为：
textfast-topic：轻量、实时任务 slow-topic：耗时、批量任务 retry-topic：失败重试任务 dead-letter-topic：人工处理任务
检查消费代码是否存在：
如果不同消息之间没有严格顺序依赖，应尽可能降低锁粒度。
即使队列整体消费能力足够，单个分区过热也会导致局部积压。
常见原因：
优化方式：
例如：
text原分区 Key： tenantId 优化后： tenantId + hash(orderId) % 8
需要注意：增加随机维度可能破坏同一 Key 的严格顺序。
消息体积变大会影响：
建议：
推荐模式：
json{
"eventId": "evt_123",
"bizId": "order_456",
"eventType": "ORDER_PAID",
"occurredAt": "2026-07-15T10:00:00Z",
"dataVersion": 3
}
消费者根据 bizId 按需读取完整业务数据，而不是在消息中传递完整对象。
积压中可能存在大量重复消息。
常见原因：
建议通过以下字段实现幂等：
texteventId requestId businessId + eventType + version
典型幂等方式：
SET NX注意：
消息队列通常只能保证至少一次投递，业务系统必须具备幂等处理能力。
严格顺序会显著限制并发能力。
需要明确：
推荐优先使用局部顺序：
text全局顺序 ↓ 租户级顺序 ↓ 用户级顺序 ↓ 订单级顺序
顺序范围越小，可用并发度越高。
markdown## 消息积压根因分析
### 1. 事件概述
- 发生时间：
- 恢复时间：
- 影响时长：
- 影响业务：
- 影响消息量：
- 最大消息延迟：
- 是否造成数据丢失：
- 是否造成重复处理：
### 2. 现象
- 生产速率：
- 消费速率：
- 峰值积压量：
- 消费失败率：
- 最老消息年龄：
- 异常消费者数量：
- 下游异常情况：
### 3. 直接原因
例如：
- 数据库连接池耗尽
- 消费代码发布后单条处理耗时从 20ms 上升到 500ms
- 上游全量补偿任务产生突发流量
- 单条异常消息无限重试
- 消费组频繁重平衡
### 4. 根本原因
例如：
- 缺少容量评估
- 缺少消费延迟告警
- 重试机制设计不合理
- 缺少死信队列
- 发布前未进行压测
- 上下游缺少流量保护
- 消费逻辑与外部依赖强耦合
### 5. 应急处理
- 限流措施：
- 扩容措施：
- 回滚措施：
- 降级措施：
- 消息隔离措施：
- 数据补偿措施：
### 6. 改进项
| 改进项 | 负责人 | 截止时间 | 状态 |
|---|---|---|---|
| 增加消费延迟告警 | | | |
| 增加死信队列 | | | |
| 优化批量写入 | | | |
| 完善容量压测 | | | |
消息队列不能只监控“消息数量”，应建立覆盖生产、存储、消费和业务结果的指标体系。
仅根据积压条数告警可能不准确。
例如：
text积压 100 万条消息
如果消费速度很快，可能几分钟即可处理完。
而：
text积压 1000 条消息
如果最老消息已经等待 2 小时，则可能已经严重影响业务。
因此建议重点监控：
text最老未消费消息年龄
告警示例：
| 告警级别 | 条件 |
|---|---|
| 提示 | 最老消息年龄 > 1 分钟 |
| 警告 | 最老消息年龄 > 5 分钟 |
| 严重 | 最老消息年龄 > 15 分钟 |
| 紧急 | 最老消息年龄 > 30 分钟 |
实际阈值应根据业务 SLA 制定。
避免积压达到严重程度后才告警。
推荐设置：
text一级告警：消费速率连续下降 二级告警：生产速率持续高于消费速率 三级告警：积压量达到容量阈值 四级告警：消息年龄超过业务 SLA 五级告警：消息即将过期或磁盘即将耗尽
告警内容应包含：
系统设计时应评估：
text峰值生产速率 平均生产速率 单实例消费速率 消费者最大扩容数 下游最大承载能力 允许的最大消息延迟 最大可接受积压量 消息保留时长 故障恢复时间
text所需消费者数量 = 峰值生产速率 ÷ 单实例稳定消费速率 × 安全系数
示例：
text峰值生产速率：20,000 条/秒 单实例稳定消费速率：2,000 条/秒 安全系数：1.5 所需实例数： 20,000 ÷ 2,000 × 1.5 = 15
建议至少准备 15 个实例的容量。
text最大积压量 = 允许的最大恢复时间 × 净消费速率
例如：
text要求 30 分钟内清理积压 净消费速率为 5,000 条/秒 最大积压量： 30 × 60 × 5,000 = 9,000,000 条
消费者不应长期运行在接近满负载状态。
建议正常运行时：
text平均资源使用率控制在 50%～70%
保留余量用于应对：
如果消费者长期需要 90% 以上资源才能跟上生产速度，说明系统没有足够安全余量。
可根据以下指标自动扩容：
推荐扩容条件：
text最老消息年龄持续升高 且 消费速率低于生产速率 且 下游系统仍有剩余容量
缩容时应避免频繁抖动，可设置：
注意：
自动扩容必须与分区数量和下游容量联动，否则扩容可能无效，甚至放大故障。
生产端应具备：
消费端应具备：
Broker 侧应具备：
建议将重试与主消费队列分离。
推荐架构：
text主队列 → 消费失败 → 1 分钟重试队列 → 5 分钟重试队列 → 30 分钟重试队列 → 死信队列
每次失败应记录：
json{
"eventId": "evt_123",
"retryCount": 3,
"firstFailedAt": "2026-07-15T10:00:00Z",
"lastFailedAt": "2026-07-15T10:30:00Z",
"failureCode": "DATABASE_TIMEOUT",
"failureMessage": "database connection timeout"
}
重试必须满足：
消息应明确以下生命周期状态：
text已生产 → 已存储 → 已拉取 → 处理中 → 处理成功 → 处理失败 → 等待重试 → 死信 → 已补偿
同时明确：
建议定期验证：
可演练以下故障：
演练目标不是“系统完全不出问题”，而是验证：
消费代码变更可能导致单条处理耗时明显增加。
发布前应对比：
text变更前平均耗时 变更后平均耗时 变更前 P99 变更后 P99 变更前吞吐量 变更后吞吐量 数据库调用次数变化 外部接口调用次数变化 消息体积变化 内存占用变化
可设置发布门禁：
text单条消费耗时增长超过 20% 或 吞吐量下降超过 15% 或 错误率超过阈值 → 阻止发布
应急预案至少应包含：
markdown# 消息队列积压应急预案
## 基本信息
- 队列名称：
- 消费组名称：
- 业务负责人：
- 技术负责人：
- 消息中间件负责人：
- 下游系统负责人：
## 核心指标
- 正常生产速率：
- 峰值生产速率：
- 单实例消费速率：
- 正常消费者数量：
- 最大消费者数量：
- 消息延迟 SLA：
- 消息保留时长：
## 应急操作
1. 检查消费者存活状态
2. 检查最近发布和配置变更
3. 检查数据库、缓存和外部接口
4. 判断是否需要生产端限流
5. 判断是否需要回滚
6. 判断是否需要消费者扩容
7. 判断是否需要隔离异常消息
8. 判断是否需要暂停非核心逻辑
9. 评估积压清理时间
10. 持续观察下游系统负载
## 禁止操作
- 未确认下游容量前无限扩容消费者
- 未进行幂等校验前重复回放消息
- 直接删除核心业务消息
- 对异常消息进行无限重试
- 在没有备份的情况下重置消费位点
- 同时进行多项高风险变更
text发现消息积压 │ ▼ 确认生产速率、消费速率、消息年龄 │ ▼ 消费端是否故障？ ┌────┴────┐ │ │ 是 否 │ │ 恢复消费者 判断生产是否突增 回滚版本 │ 修复依赖 ┌───┴───┐ │ │ │ │ 是 否 │ │ │ │ 限流/暂停 分析消费瓶颈 │ 非核心流量 │ └──────────┬───────────┘ ▼ 是否存在异常消息？ ┌────┴────┐ │ │ 是 否 │ │ 隔离到死信队列 继续分析 │ │ └────┬────┘ ▼ 下游是否有扩容余量？ ┌────┴────┐ │ │ 是 否 │ │ 扩容消费者 降级消费逻辑 增加并发 限速处理 │ │ └────┬────┘ ▼ 批量消费、批量写入 │ ▼ 实时消息与历史消息分流 │ ▼ 持续计算预计清理时间 │ ▼ 积压恢复至正常水平 │ ▼ 根因分析与长期整改
核心目标：
阻止积压继续扩大，并保障核心业务。
关键动作：
核心目标：
找到消费吞吐下降或生产流量异常的真实原因。
关键动作：
核心目标：
让积压能够提前发现、自动控制并快速恢复。
关键动作：
消息队列积压处理的正确顺序是：先限流止血、恢复消费并隔离异常消息，再定位生产与消费链路瓶颈，最后通过容量治理、监控告警、分级重试、自动扩缩容和故障演练建立长期防线。
本文作者:曹子昂
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！