凌晨两点十七分我手机上的告警通道开始连续发声。监控面板显示order_process队列的 Ready 消息数在十分钟内从 200 冲到了 5000而消费者进程明明还活着日志里却在疯狂刷同一条消息的消费失败栈——又是那台订单处理服务。做 Java 后端这些年RabbitMQ 的坑我踩过不少但最让人头疼的永远是这一类消息积压、消费卡死、失败消息无限循环重试。这篇文章我不打算讲太多 RabbitMQ 的基础概念而是把消费者消息限流、超时、死信队列这三件事串成一条完整的问题链路讲清楚各自的原理、配置方式以及最后落到若依框架里时如何把自带的那套能用但不可靠的 RabbitMQ 集成升级成真正抗得住的方案。适合正在用 RabbitMQ 做异步业务、被消费端问题折磨过、以及在使用若依做快速开发但想补上中间件短板的工程师。1. 先看一个真实事故消息积压让值班手机响了一夜1.1 事故现场的三个关键现象那次线上事故我给团队复盘过很多次。表象有三个。第一个是队列监控里messages_ready飙升messages_unacknowledged也不低第二个是消费者服务 CPU 占用不高、内存正常但线程 Dump 出来发现有一批线程阻塞在数据库连接等待上第三个是日志里能看到同一条消息被消费了十几次每次都是同样地抛异常然后又同样地被重新放回队列。这三个现象凑在一起基本可以还原全貌。RabbitMQ 默认情况下对消费者的投递是没有数量限制的——消息会尽可能快地全部推给消费者进程。消费者拿到一批消息后开始处理结果处理逻辑里有一段本地资源的竞争导致线程卡住这批消息就一直处于 Unacked 状态既不确认也不拒绝。而队列头部那条失败的消息因为反复被 requeue把整条队列的消费顺序卡住了后面的消息全部堵在它后面形成了队头阻塞。结论很简单消费端没有限流没有超时兜底失败的消息也没有终点站。1.2 根因链条限流、超时、死信缺一不可那次事故之后我总结出一个观点RabbitMQ 消费端的高可靠性并不是靠某一个参数而是靠一条完整的链路。限流负责让消费者按自己的处理能力拉取消息避免一次性撑爆本地缓冲超时负责识别那些连 ACK 都回不来的僵尸消息让卡死的消费线程被重置死信队列负责收容所有确认失败、TTL 过期、队列超限的消息让异常数据有地方去而不是堵在业务队列里无限循环。这三者缺一个另外两个的威力都会大打折扣。只有限流没有超时消费者本地堆积的消息多了之后RabbitMQ 服务端仍然认为它健在问题被藏在消费进程内部只有超时没有死信消息被拒绝后还是会重新投递循环重试会把真正有问题的数据反复放大。所以下面我按这条链路一步步拆每一步都给到可以直接落地的配置和代码。2. 消费者限流prefetch参数才是公平调度的根本2.1 QoS机制与prefetch的工作原理RabbitMQ 的限流机制底层叫 Quality of ServiceQoS。它的核心是给每个消费者设置一个未确认消息窗口prefetchCount。简单说消费者在收到 ACK 之前最多能从服务端预取多少条消息。这个窗口一旦填满服务端就不会再向这个消费者推送新消息直到消费者确认了其中的某一条。这里有个容易混淆的点RabbitMQ 推消息给消费者和消费者取消息在外层接口看起来都叫消费但底层是服务端主动推送push模式。所以 QoS 限流不是在消费者这边取少一点而是告诉服务端你最多同时给我这么多多了我处理不过来。把 RabbitMQ 想象成一个打饭窗口prefetch 就是这个窗口一次最多能递出来的餐盘数。餐盘递多了你手上端不过来后面的菜就只能搁在台面上等。2.2 Spring AMQP里的限流配置与手动basicQos在 Spring Boot 项目里限流的配置大多数情况下不需要手动操作 Channel只要改application.yml里的监听器参数spring: rabbitmq: listener: simple: prefetch: 100 acknowledge-mode: AUTO concurrent-consumers: 5 max-concurrent-consumers: 15这段配置定义了每个消费者线程的预取窗口是 100 条整个监听器容器初始 5 个消费者最多扩展到 15 个。也就是说全站积压最严重时单队列未被确认的消息数理论上限是 15 乘 100也就是 1500 条左右。这个数字是可控的。如果你是手动创建 Channel 做底层消费那就直接调用basicQosChannel channel connection.createChannel(); channel.basicQos(100); String consumerTag channel.basicConsume(order_process, false, consumer);注意手动模式下第二个参数要传false表示关闭自动 ACK由你自己控制确认时机。手动 ACK 配合限流才是完整的一套。如果打开自动 ACK限流就形同虚设了——消息刚推过来就被标记确认RabbitMQ 没有任何压力反馈Buffer 里的消息则可能无限堆积。2.3 prefetch取值经验别盲目设1也别放任默认关于 prefetch 的大小业界有几种倾向。设成 1 是最稳妥的每条消息确认后再拉下一条消息永远不会在消费者本地积压。但代价是吞吐量非常难看尤其在业务处理只有几十毫秒的场景下网络 RTT 反而成了瓶颈。我实测过同样一批 10 万条消息prefetch1 时消费完成时间比 prefetch100 慢了将近三倍。反过来说完全不设限流默认行为更危险。消费者本地缓冲被塞满后一旦处理逻辑出问题积压发生在你进程内部监控根本看不出来。我现在的经验是普通业务处理在几百毫秒级别的prefetch 取 50 到 200 之间比较合适如果单条消费涉及远程调用、响应时间波动大取 30 到 50如果消费逻辑极轻、纯内存计算可以拉到 500但一定要配合超时兜底。判断标准很简单prefetch 乘以单条处理时间的均值应该接近你期望的消费者单线程吞吐延迟的一倍左右给缓冲留足空间。还有一个值得注意的点限流参数要和并发消费者一起调。不要只改 prefetch 不调并发。理想状态是消费者总数乘以 prefetch 的积略大于生产端的瞬时峰值否则消费速度天然就跟不上生产速度队列积压只是时间问题。3. 消息超时TTL与消费者交付超时的边界3.1 消息存活时间TTL队列级与消息级限流解决的是消费者拉太多的问题而超时解决的是消息在某个环节停留太久的问题。RabbitMQ 的 TTLTime To Live可以作用在两个层面。队列级别通过在声明队列时加参数实现Bean public Queue orderProcessQueue() { return QueueBuilder.durable(order_process) .withArgument(x-message-ttl, 300000) .withArgument(x-dead-letter-exchange, dlx.main) .withArgument(x-dead-letter-routing-key, dlx.main.order) .build(); }消息级别则是在发布时设置expiration属性MessageProperties props new MessageProperties(); props.setExpiration(300000); Message message new Message(payload, props); rabbitTemplate.send(order_exchange, order.process, message);队列级 TTL 是统一管理所有进入该队列的消息都适用消息级 TTL 是单条控制更灵活。两者都设置时会取较小值生效。这里有个实际切身体会如果只是想把卡在队列里没人消费的旧数据清掉建议优先用队列级 TTL这样运维上不需要依赖生产端每条消息都记得设置过期时间。TTL 到期后的消息不会自动消失而是会被转投到死信交换机——这正好和后面的死信队列设计衔接上。3.2 消费者交付超时兜住卡死的消费线程除了消息本身的 TTL还有一类超时是针对消费者交付过程的。Spring AMQP 在较新的版本里提供了consumer-timeout参数spring: rabbitmq: listener: simple: consumer-timeout: 60000这个参数的含义是消费者从队列拉取了一条消息后如果在超时时间内没有返回没有 ACK/NACKRabbitMQ 客户端会认为该消费者已经不可用主动关闭这个通道并把消息重新投递给其他消费者。它兜住的是那些线程卡死、连接还活着的场景——比如消费者内部等待某个分布式锁、数据库连接池耗尽、或者长时间 Full GC。没有这个参数时RabbitMQ 服务端唯一的判断依据是 TCP 连接是否存活只要连接不关它就认为消费者一切正常消息就永远卡在 Unacked 状态。3.3 重试与超时的配合本地重试外层兜底Spring AMQP 内置的重试机制是消费端本地的。配置是这样的spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 1.0 max-interval: 10000这段配置表示业务代码抛出异常时Spring 会在消费端本地按 1 秒间隔最多重试 3 次。注意这里的重试是本地重试消息并不会被退回 RabbitMQ只有本地重试次数耗尽后异常才会进入监听器容器处理逻辑由default-requeue-rejected决定是否重新入队。我把超时机制总结成一张层次表方便区分机制作用层面触发条件兜住的问题消息 TTL队列/消息消息在队列存活超过指定时间没人消费的旧数据队列长度限制队列消息数量超过x-max-length队列无限膨胀消费者交付超时连接/通道消费者拉取消息后超时未返回消费线程卡死Spring 本地重试应用内业务代码抛出异常偶发瞬时错误单看任何一层的命中场景都是小概率事件但线上把这几层叠加后整个消息链路才真正有边界感。4. 死信队列让失败的消息有终点而不是无限循环4.1 消息进入死信的三种触发条件死信队列Dead Letter Queue是 RabbitMQ 给失败消息准备的收容站。消息进入死信交换机有三种常规触发条件。第一种是消息被消费者主动拒绝。这里要区分两个参数basicReject和basicNack都支持requeue参数只有当requeuefalse时被拒绝的消息才会进入死信。第二种是消息 TTL 过期前面已经说过过期消息会被转投到死信交换机。第三种是队列达到最大长度如果队列设置了x-max-length或x-max-length-bytes溢出部分的旧消息会按规则进入死信。这三种条件覆盖了大多数消息不该再留在业务队列里的场景。但要注意消息进入死信队列后会保留原始内容、原始 routingKey、以及一部分头信息死信消费者可以根据原始信息做进一步处理比如告警、人工介入、落库标记等。4.2 统一死信交换机与队列声明我建议在一个项目里使用统一的死信交换机而不是每个业务队列都建一套自己的死信链路。统一的好处是监控好做一个交换机一条队列从管理台一眼就能看出有多少消息走到了失败终点。下面是一组完整的声明代码Configuration public class RabbitDlxConfig { public static final String DLX_EXCHANGE dlx.main; public static final String DLX_QUEUE dlx.main.queue; public static final String DLX_ROUTING_KEY dlx.main; Bean public DirectExchange dlxExchange() { return new DirectExchange(DLX_EXCHANGE, true, false); } Bean public Queue dlxQueue() { return QueueBuilder.durable(DLX_QUEUE) .withArgument(x-message-ttl, 3600000) .build(); } Bean public Binding dlxBinding() { return BindingBuilder.bind(dlxQueue()) .to(dlxExchange()) .with(DLX_ROUTING_KEY); } Bean public Queue orderProcessQueue() { return QueueBuilder.durable(order_process) .withArgument(x-dead-letter-exchange, DLX_EXCHANGE) .withArgument(x-dead-letter-routing-key, DLX_ROUTING_KEY) .withArgument(x-message-ttl, 300000) .build(); } }这里有两个经验。第一不要把业务队列和死信队列设成同一个 TTL 策略。业务队列的 TTL 是给正常消费一个时限死信队列的 TTL 则是防止死信无限堆积——没人处理的死信也得有个出口。第二routingKey 的设计不要只用一个平铺的名字建议带上业务标识比如上面的dlx.main是对应所有业务如果你希望按业务细分死信可以改成dlx.main.order、dlx.main.payment然后让业务队列声明时使用各自的 key。4.3 为什么必须关闭requeue队头阻塞的真相默认情况下basicNack里requeue如果传true消息会被放回原队列头部。这是个非常隐蔽的坑。表面上看消息回到队列后还能再次被消费似乎无害但实际上它会立刻变成队头阻塞问题的源头。原因在于 RabbitMQ 的消息投递是按照队列顺序走的一条被 requeue 的消息通常还在消费者本地缓存或服务端待投递缓冲区里它会排在所有新消息前面。如果这条消息每次都处理失败、每次都返队那全队列的消费进度都会被它卡住。我在第一节复盘的那次事故核心就是这个问题。所以要养成一个习惯业务侧重试逻辑放到应用内部Spring 的本地重试机制一旦本地重试次数耗尽对外层明确basicNack且requeuefalse让消息走死信。这也是为什么配置文件里我要把default-requeue-rejected设成false。spring: rabbitmq: listener: simple: default-requeue-rejected: false这个配置配合本地重试以后消息的生命周期就变成正常处理成功 - ACK 消失处理失败 - 应用内重试 - 仍失败 - 拒绝且不回队 - 进入死信队列。整个过程没有无限循环没有队头阻塞每条失败消息最多在业务队列里存在一个有限的时间。5. 若依框架的RabbitMQ集成升级从能用变成可靠5.1 若依自带的Rabbit配置到底差在哪若依RuoYi作为国内使用率很高的快速开发脚手架业务模块生成效率确实高但它的消息中间件集成停留在能用阶段。很多从若依二开的项目RabbitConfig长得很类似手动 new 一个CachingConnectionFactory塞几个连接参数然后给RabbitTemplate配上Jackson2JsonMessageConverter就完事了。监听器直接用RabbitListener靠 Spring Boot 默认的监听器容器跑。这套配置的问题在线上放大得很明显。第一没有开启 publisher confirm 和 returns发消息发丢了发送方完全无感知第二没有配置 prefetch 和并发消费者消费吞吐全凭默认值第三没有设置任何死信相关参数一旦消息消费失败就是无脑 requeue第四Jackson2JsonMessageConverter只装了发送侧如果发送对象和消费对象类型不一致反序列化也会成为新的坑点。5.2 升级后的ConnectionFactory与RabbitTemplate升级的第一步是重写连接工厂和模板把发送方确认、路由失败回调、消息转换器全部补齐。Configuration public class RabbitReliableConfig { Value(${spring.rabbitmq.host}) private String host; Value(${spring.rabbitmq.port}) private Integer port; Value(${spring.rabbitmq.username}) private String username; Value(${spring.rabbitmq.password}) private String password; Bean public ConnectionFactory rabbitConnectionFactory() { CachingConnectionFactory factory new CachingConnectionFactory(); factory.setHost(host); factory.setPort(port); factory.setUsername(username); factory.setPassword(password); // 开启发送方确认模式使用 CorrelationData 关联每条消息 factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED); // 开启路由失败返回 factory.setPublisherReturns(true); return factory; } Bean public Jackson2JsonMessageConverter jacksonMessageConverter() { return new Jackson2JsonMessageConverter(); } Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Jackson2JsonMessageConverter messageConverter) { RabbitTemplate template new RabbitTemplate(connectionFactory); template.setMessageConverter(messageConverter); template.setMandatory(true); template.setConfirmCallback((correlationData, ack, cause) - { if (!ack) { log.error(消息发送确认失败, cause: {}, cause); } }); template.setReturnsCallback(returned - { log.error(消息路由失败, 交换机: {}, 路由键: {}, 应答文本: {}, returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()); }); return template; } }这里的mandatorytrue加 returns 回调解决的是消息发到交换机但路由不到队列的场景。以前被问得最多的问题是rabbitTemplate.convertAndSend返回时没抛异常是不是就代表发送成功了不是。只要消息到了交换机convertAndSend就不抛异常但如果交换机没有绑定的队列这条消息就被丢了。开了 mandatory 和 returns 之后这种情况会回调回来逻辑里可以接告警或者转人工。5.3 监听器容器工厂与全局异常兜底发送侧补强之后消费侧同样要有一个显式的监听器容器工厂否则 Spring Boot 默认容器不会读取你项目里自定义的 QoS、并发、重试参数。Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( ConnectionFactory connectionFactory, Jackson2JsonMessageConverter messageConverter) { SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(messageConverter); factory.setPrefetchCount(100); factory.setConcurrentConsumers(5); factory.setMaxConcurrentConsumers(15); factory.setAcknowledgeMode(AcknowledgeMode.AUTO); factory.setDefaultRequeueRejected(false); factory.setErrorHandler(new GlobalRabbitErrorHandler()); return factory; }这里要特意说明AcknowledgeMode.AUTO和defaultRequeueRejectedfalse的配合逻辑。AUTO 模式下RabbitListener方法正常返回就自动 ACK方法抛异常就自动 NACK。配合defaultRequeueRejectedfalseNACK 时不回队消息直接进死信。于是你业务代码里不需要手工去 catch 异常再调用basicNackSpring 已经帮你把这条链路闭环了。需要注意的是默认情况下若依项目里如果存在多个监听器容器最好把RabbitListener上的containerFactory属性显式指到这个新建的工厂防止部分监听器落到默认配置上。全局异常处理可以用一个实现RabbitListenerErrorHandler的组件Component public class GlobalRabbitErrorHandler implements RabbitListenerErrorHandler { private static final Logger log LoggerFactory.getLogger(GlobalRabbitErrorHandler.class); Override public Object handleError(Message amqpMessage, org.springframework.messaging.Message? message, ListenerExecutionFailedException exception) { log.error(RabbitMQ 消费异常, queue: {}, message: {}, amqpMessage.getMessageProperties().getConsumerQueue(), new String(amqpMessage.getBody()), exception); // 返回 null配合 defaultRequeueRejectedfalse 使消息进入死信 return null; } }在这个异常处理器里可以接告警、落库、记指标总之失败消息在进入死信之前至少有一个地方统一留下痕迹。等到死信消费者再读到这条消息时就能结合这里的日志做业务侧补偿。6. 升级完怎么验证一张表看清限流、超时、死信是否生效6.1 验证清单与操作步骤配置改完之后不上线直接信那是自己骗自己。下面是我每次升级 RabbitMQ 消费端之后必跑的验证清单照着做一遍基本心里就有底了。验证项操作步骤预期结果限流生效停掉所有消费者用脚本向队列发送 1000 条消息再启动消费者管理后台messages_unacknowledged不会超过 消费者数 乘 prefetch且趋于稳定处理超时兜底在RabbitListener方法里Thread.sleep(5分钟)观察日志达到consumer-timeout后通道被关闭消息重新投递失败消息进死信监听方法里直接抛异常且本地重试设为 3 次观察 RabbitMQ 管理台业务队列消息数不变死信队列收到 3 条对应的失败消息TTL 过期进死信给业务队列设置 5 分钟 TTL发一条消息不去消费5 分钟后消息进入死信队列业务队列清空发送失败感知故意让消息路由到不存在的队列观察应用日志returnsCallback打印消息路由失败完整信息跑这组验证的时候最实用的命令行是rabbitmqctl list_queues name messages_ready messages_unacknowledgedmessages_ready是待在队列里等待投递的消息数messages_unacknowledged是已推给消费者但还没确认的消息数。限流有没有生效看 unack 的数量能不能被 prefetch 乘消费者数限制住一目了然。6.2 长期运维的四个原则验证通过只能说上线那天是好的。真正让这套机制长期可靠靠的是运维习惯。我踩过几轮坑之后给自己定了几条规矩。第一死信队列要每天看。死信数量从零涨到几百通常不需要恐慌但要观察增长速度。如果每天都在稳定增长说明业务侧有系统性失败不是偶发抖动。第二不要为了方便把死信队列的 TTL 设成无限大。死信本身就是异常数据保留价值随时间递减建议保留 1 到 3 天到期自动清理。第三发布端确认回调里的失败日志不能只打印不处理。哪怕只是加一个简单的计数指标也要让失败可见。第四改 prefetch 或并发参数时要压测后再上线不要凭感觉拍脑袋。同样一台 4C8G 的机器prefetch 从 100 调到 300消费吞吐可能只涨不到 10%但本地缓冲里的消息积压却翻了三倍GC 压力明显变大这不划算。用命令行再查一下未确认消息数和死信队列深度这俩指标配上消费者线程数基本就能判断整个消息链路的健康状况。结尾我在实际项目里最深的一个体会是RabbitMQ 消费端的问题从来不是单点参数能解决的。prefetch 管的是消费者拉多少consumer-timeout 管的是消费线程卡了怎么办TTL 管的是消息待多久DLX 管的是失败去哪里。四者配合起来消息才会真正走完投递-处理-成功或失败的全生命周期。如果你正在用若依做快速开发搭完业务模块后不妨抽半天时间把自带那套 RabbitMQ 集成按今天的方案升级一下。这个投入比任何业务代码都值得——因为消息中间件平时不说话一出问题就是半夜值班电话连着响。当然这种事我希望你永远不要亲自经历。
凌晨两点十七分我手机上的告警通道开始连续发声。监控面板显示order_process队列的 Ready 消息数在十分钟内从 200 冲到了 5000而消费者进程明明还活着日志里却在疯狂刷同一条消息的消费失败栈——又是那台订单处理服务。做 Java 后端这些年RabbitMQ 的坑我踩过不少但最让人头疼的永远是这一类消息积压、消费卡死、失败消息无限循环重试。这篇文章我不打算讲太多 RabbitMQ 的基础概念而是把消费者消息限流、超时、死信队列这三件事串成一条完整的问题链路讲清楚各自的原理、配置方式以及最后落到若依框架里时如何把自带的那套能用但不可靠的 RabbitMQ 集成升级成真正抗得住的方案。适合正在用 RabbitMQ 做异步业务、被消费端问题折磨过、以及在使用若依做快速开发但想补上中间件短板的工程师。1. 先看一个真实事故消息积压让值班手机响了一夜1.1 事故现场的三个关键现象那次线上事故我给团队复盘过很多次。表象有三个。第一个是队列监控里messages_ready飙升messages_unacknowledged也不低第二个是消费者服务 CPU 占用不高、内存正常但线程 Dump 出来发现有一批线程阻塞在数据库连接等待上第三个是日志里能看到同一条消息被消费了十几次每次都是同样地抛异常然后又同样地被重新放回队列。这三个现象凑在一起基本可以还原全貌。RabbitMQ 默认情况下对消费者的投递是没有数量限制的——消息会尽可能快地全部推给消费者进程。消费者拿到一批消息后开始处理结果处理逻辑里有一段本地资源的竞争导致线程卡住这批消息就一直处于 Unacked 状态既不确认也不拒绝。而队列头部那条失败的消息因为反复被 requeue把整条队列的消费顺序卡住了后面的消息全部堵在它后面形成了队头阻塞。结论很简单消费端没有限流没有超时兜底失败的消息也没有终点站。1.2 根因链条限流、超时、死信缺一不可那次事故之后我总结出一个观点RabbitMQ 消费端的高可靠性并不是靠某一个参数而是靠一条完整的链路。限流负责让消费者按自己的处理能力拉取消息避免一次性撑爆本地缓冲超时负责识别那些连 ACK 都回不来的僵尸消息让卡死的消费线程被重置死信队列负责收容所有确认失败、TTL 过期、队列超限的消息让异常数据有地方去而不是堵在业务队列里无限循环。这三者缺一个另外两个的威力都会大打折扣。只有限流没有超时消费者本地堆积的消息多了之后RabbitMQ 服务端仍然认为它健在问题被藏在消费进程内部只有超时没有死信消息被拒绝后还是会重新投递循环重试会把真正有问题的数据反复放大。所以下面我按这条链路一步步拆每一步都给到可以直接落地的配置和代码。2. 消费者限流prefetch参数才是公平调度的根本2.1 QoS机制与prefetch的工作原理RabbitMQ 的限流机制底层叫 Quality of ServiceQoS。它的核心是给每个消费者设置一个未确认消息窗口prefetchCount。简单说消费者在收到 ACK 之前最多能从服务端预取多少条消息。这个窗口一旦填满服务端就不会再向这个消费者推送新消息直到消费者确认了其中的某一条。这里有个容易混淆的点RabbitMQ 推消息给消费者和消费者取消息在外层接口看起来都叫消费但底层是服务端主动推送push模式。所以 QoS 限流不是在消费者这边取少一点而是告诉服务端你最多同时给我这么多多了我处理不过来。把 RabbitMQ 想象成一个打饭窗口prefetch 就是这个窗口一次最多能递出来的餐盘数。餐盘递多了你手上端不过来后面的菜就只能搁在台面上等。2.2 Spring AMQP里的限流配置与手动basicQos在 Spring Boot 项目里限流的配置大多数情况下不需要手动操作 Channel只要改application.yml里的监听器参数spring: rabbitmq: listener: simple: prefetch: 100 acknowledge-mode: AUTO concurrent-consumers: 5 max-concurrent-consumers: 15这段配置定义了每个消费者线程的预取窗口是 100 条整个监听器容器初始 5 个消费者最多扩展到 15 个。也就是说全站积压最严重时单队列未被确认的消息数理论上限是 15 乘 100也就是 1500 条左右。这个数字是可控的。如果你是手动创建 Channel 做底层消费那就直接调用basicQosChannel channel connection.createChannel(); channel.basicQos(100); String consumerTag channel.basicConsume(order_process, false, consumer);注意手动模式下第二个参数要传false表示关闭自动 ACK由你自己控制确认时机。手动 ACK 配合限流才是完整的一套。如果打开自动 ACK限流就形同虚设了——消息刚推过来就被标记确认RabbitMQ 没有任何压力反馈Buffer 里的消息则可能无限堆积。2.3 prefetch取值经验别盲目设1也别放任默认关于 prefetch 的大小业界有几种倾向。设成 1 是最稳妥的每条消息确认后再拉下一条消息永远不会在消费者本地积压。但代价是吞吐量非常难看尤其在业务处理只有几十毫秒的场景下网络 RTT 反而成了瓶颈。我实测过同样一批 10 万条消息prefetch1 时消费完成时间比 prefetch100 慢了将近三倍。反过来说完全不设限流默认行为更危险。消费者本地缓冲被塞满后一旦处理逻辑出问题积压发生在你进程内部监控根本看不出来。我现在的经验是普通业务处理在几百毫秒级别的prefetch 取 50 到 200 之间比较合适如果单条消费涉及远程调用、响应时间波动大取 30 到 50如果消费逻辑极轻、纯内存计算可以拉到 500但一定要配合超时兜底。判断标准很简单prefetch 乘以单条处理时间的均值应该接近你期望的消费者单线程吞吐延迟的一倍左右给缓冲留足空间。还有一个值得注意的点限流参数要和并发消费者一起调。不要只改 prefetch 不调并发。理想状态是消费者总数乘以 prefetch 的积略大于生产端的瞬时峰值否则消费速度天然就跟不上生产速度队列积压只是时间问题。3. 消息超时TTL与消费者交付超时的边界3.1 消息存活时间TTL队列级与消息级限流解决的是消费者拉太多的问题而超时解决的是消息在某个环节停留太久的问题。RabbitMQ 的 TTLTime To Live可以作用在两个层面。队列级别通过在声明队列时加参数实现Bean public Queue orderProcessQueue() { return QueueBuilder.durable(order_process) .withArgument(x-message-ttl, 300000) .withArgument(x-dead-letter-exchange, dlx.main) .withArgument(x-dead-letter-routing-key, dlx.main.order) .build(); }消息级别则是在发布时设置expiration属性MessageProperties props new MessageProperties(); props.setExpiration(300000); Message message new Message(payload, props); rabbitTemplate.send(order_exchange, order.process, message);队列级 TTL 是统一管理所有进入该队列的消息都适用消息级 TTL 是单条控制更灵活。两者都设置时会取较小值生效。这里有个实际切身体会如果只是想把卡在队列里没人消费的旧数据清掉建议优先用队列级 TTL这样运维上不需要依赖生产端每条消息都记得设置过期时间。TTL 到期后的消息不会自动消失而是会被转投到死信交换机——这正好和后面的死信队列设计衔接上。3.2 消费者交付超时兜住卡死的消费线程除了消息本身的 TTL还有一类超时是针对消费者交付过程的。Spring AMQP 在较新的版本里提供了consumer-timeout参数spring: rabbitmq: listener: simple: consumer-timeout: 60000这个参数的含义是消费者从队列拉取了一条消息后如果在超时时间内没有返回没有 ACK/NACKRabbitMQ 客户端会认为该消费者已经不可用主动关闭这个通道并把消息重新投递给其他消费者。它兜住的是那些线程卡死、连接还活着的场景——比如消费者内部等待某个分布式锁、数据库连接池耗尽、或者长时间 Full GC。没有这个参数时RabbitMQ 服务端唯一的判断依据是 TCP 连接是否存活只要连接不关它就认为消费者一切正常消息就永远卡在 Unacked 状态。3.3 重试与超时的配合本地重试外层兜底Spring AMQP 内置的重试机制是消费端本地的。配置是这样的spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 initial-interval: 1000 multiplier: 1.0 max-interval: 10000这段配置表示业务代码抛出异常时Spring 会在消费端本地按 1 秒间隔最多重试 3 次。注意这里的重试是本地重试消息并不会被退回 RabbitMQ只有本地重试次数耗尽后异常才会进入监听器容器处理逻辑由default-requeue-rejected决定是否重新入队。我把超时机制总结成一张层次表方便区分机制作用层面触发条件兜住的问题消息 TTL队列/消息消息在队列存活超过指定时间没人消费的旧数据队列长度限制队列消息数量超过x-max-length队列无限膨胀消费者交付超时连接/通道消费者拉取消息后超时未返回消费线程卡死Spring 本地重试应用内业务代码抛出异常偶发瞬时错误单看任何一层的命中场景都是小概率事件但线上把这几层叠加后整个消息链路才真正有边界感。4. 死信队列让失败的消息有终点而不是无限循环4.1 消息进入死信的三种触发条件死信队列Dead Letter Queue是 RabbitMQ 给失败消息准备的收容站。消息进入死信交换机有三种常规触发条件。第一种是消息被消费者主动拒绝。这里要区分两个参数basicReject和basicNack都支持requeue参数只有当requeuefalse时被拒绝的消息才会进入死信。第二种是消息 TTL 过期前面已经说过过期消息会被转投到死信交换机。第三种是队列达到最大长度如果队列设置了x-max-length或x-max-length-bytes溢出部分的旧消息会按规则进入死信。这三种条件覆盖了大多数消息不该再留在业务队列里的场景。但要注意消息进入死信队列后会保留原始内容、原始 routingKey、以及一部分头信息死信消费者可以根据原始信息做进一步处理比如告警、人工介入、落库标记等。4.2 统一死信交换机与队列声明我建议在一个项目里使用统一的死信交换机而不是每个业务队列都建一套自己的死信链路。统一的好处是监控好做一个交换机一条队列从管理台一眼就能看出有多少消息走到了失败终点。下面是一组完整的声明代码Configuration public class RabbitDlxConfig { public static final String DLX_EXCHANGE dlx.main; public static final String DLX_QUEUE dlx.main.queue; public static final String DLX_ROUTING_KEY dlx.main; Bean public DirectExchange dlxExchange() { return new DirectExchange(DLX_EXCHANGE, true, false); } Bean public Queue dlxQueue() { return QueueBuilder.durable(DLX_QUEUE) .withArgument(x-message-ttl, 3600000) .build(); } Bean public Binding dlxBinding() { return BindingBuilder.bind(dlxQueue()) .to(dlxExchange()) .with(DLX_ROUTING_KEY); } Bean public Queue orderProcessQueue() { return QueueBuilder.durable(order_process) .withArgument(x-dead-letter-exchange, DLX_EXCHANGE) .withArgument(x-dead-letter-routing-key, DLX_ROUTING_KEY) .withArgument(x-message-ttl, 300000) .build(); } }这里有两个经验。第一不要把业务队列和死信队列设成同一个 TTL 策略。业务队列的 TTL 是给正常消费一个时限死信队列的 TTL 则是防止死信无限堆积——没人处理的死信也得有个出口。第二routingKey 的设计不要只用一个平铺的名字建议带上业务标识比如上面的dlx.main是对应所有业务如果你希望按业务细分死信可以改成dlx.main.order、dlx.main.payment然后让业务队列声明时使用各自的 key。4.3 为什么必须关闭requeue队头阻塞的真相默认情况下basicNack里requeue如果传true消息会被放回原队列头部。这是个非常隐蔽的坑。表面上看消息回到队列后还能再次被消费似乎无害但实际上它会立刻变成队头阻塞问题的源头。原因在于 RabbitMQ 的消息投递是按照队列顺序走的一条被 requeue 的消息通常还在消费者本地缓存或服务端待投递缓冲区里它会排在所有新消息前面。如果这条消息每次都处理失败、每次都返队那全队列的消费进度都会被它卡住。我在第一节复盘的那次事故核心就是这个问题。所以要养成一个习惯业务侧重试逻辑放到应用内部Spring 的本地重试机制一旦本地重试次数耗尽对外层明确basicNack且requeuefalse让消息走死信。这也是为什么配置文件里我要把default-requeue-rejected设成false。spring: rabbitmq: listener: simple: default-requeue-rejected: false这个配置配合本地重试以后消息的生命周期就变成正常处理成功 - ACK 消失处理失败 - 应用内重试 - 仍失败 - 拒绝且不回队 - 进入死信队列。整个过程没有无限循环没有队头阻塞每条失败消息最多在业务队列里存在一个有限的时间。5. 若依框架的RabbitMQ集成升级从能用变成可靠5.1 若依自带的Rabbit配置到底差在哪若依RuoYi作为国内使用率很高的快速开发脚手架业务模块生成效率确实高但它的消息中间件集成停留在能用阶段。很多从若依二开的项目RabbitConfig长得很类似手动 new 一个CachingConnectionFactory塞几个连接参数然后给RabbitTemplate配上Jackson2JsonMessageConverter就完事了。监听器直接用RabbitListener靠 Spring Boot 默认的监听器容器跑。这套配置的问题在线上放大得很明显。第一没有开启 publisher confirm 和 returns发消息发丢了发送方完全无感知第二没有配置 prefetch 和并发消费者消费吞吐全凭默认值第三没有设置任何死信相关参数一旦消息消费失败就是无脑 requeue第四Jackson2JsonMessageConverter只装了发送侧如果发送对象和消费对象类型不一致反序列化也会成为新的坑点。5.2 升级后的ConnectionFactory与RabbitTemplate升级的第一步是重写连接工厂和模板把发送方确认、路由失败回调、消息转换器全部补齐。Configuration public class RabbitReliableConfig { Value(${spring.rabbitmq.host}) private String host; Value(${spring.rabbitmq.port}) private Integer port; Value(${spring.rabbitmq.username}) private String username; Value(${spring.rabbitmq.password}) private String password; Bean public ConnectionFactory rabbitConnectionFactory() { CachingConnectionFactory factory new CachingConnectionFactory(); factory.setHost(host); factory.setPort(port); factory.setUsername(username); factory.setPassword(password); // 开启发送方确认模式使用 CorrelationData 关联每条消息 factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED); // 开启路由失败返回 factory.setPublisherReturns(true); return factory; } Bean public Jackson2JsonMessageConverter jacksonMessageConverter() { return new Jackson2JsonMessageConverter(); } Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Jackson2JsonMessageConverter messageConverter) { RabbitTemplate template new RabbitTemplate(connectionFactory); template.setMessageConverter(messageConverter); template.setMandatory(true); template.setConfirmCallback((correlationData, ack, cause) - { if (!ack) { log.error(消息发送确认失败, cause: {}, cause); } }); template.setReturnsCallback(returned - { log.error(消息路由失败, 交换机: {}, 路由键: {}, 应答文本: {}, returned.getExchange(), returned.getRoutingKey(), returned.getReplyText()); }); return template; } }这里的mandatorytrue加 returns 回调解决的是消息发到交换机但路由不到队列的场景。以前被问得最多的问题是rabbitTemplate.convertAndSend返回时没抛异常是不是就代表发送成功了不是。只要消息到了交换机convertAndSend就不抛异常但如果交换机没有绑定的队列这条消息就被丢了。开了 mandatory 和 returns 之后这种情况会回调回来逻辑里可以接告警或者转人工。5.3 监听器容器工厂与全局异常兜底发送侧补强之后消费侧同样要有一个显式的监听器容器工厂否则 Spring Boot 默认容器不会读取你项目里自定义的 QoS、并发、重试参数。Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( ConnectionFactory connectionFactory, Jackson2JsonMessageConverter messageConverter) { SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(messageConverter); factory.setPrefetchCount(100); factory.setConcurrentConsumers(5); factory.setMaxConcurrentConsumers(15); factory.setAcknowledgeMode(AcknowledgeMode.AUTO); factory.setDefaultRequeueRejected(false); factory.setErrorHandler(new GlobalRabbitErrorHandler()); return factory; }这里要特意说明AcknowledgeMode.AUTO和defaultRequeueRejectedfalse的配合逻辑。AUTO 模式下RabbitListener方法正常返回就自动 ACK方法抛异常就自动 NACK。配合defaultRequeueRejectedfalseNACK 时不回队消息直接进死信。于是你业务代码里不需要手工去 catch 异常再调用basicNackSpring 已经帮你把这条链路闭环了。需要注意的是默认情况下若依项目里如果存在多个监听器容器最好把RabbitListener上的containerFactory属性显式指到这个新建的工厂防止部分监听器落到默认配置上。全局异常处理可以用一个实现RabbitListenerErrorHandler的组件Component public class GlobalRabbitErrorHandler implements RabbitListenerErrorHandler { private static final Logger log LoggerFactory.getLogger(GlobalRabbitErrorHandler.class); Override public Object handleError(Message amqpMessage, org.springframework.messaging.Message? message, ListenerExecutionFailedException exception) { log.error(RabbitMQ 消费异常, queue: {}, message: {}, amqpMessage.getMessageProperties().getConsumerQueue(), new String(amqpMessage.getBody()), exception); // 返回 null配合 defaultRequeueRejectedfalse 使消息进入死信 return null; } }在这个异常处理器里可以接告警、落库、记指标总之失败消息在进入死信之前至少有一个地方统一留下痕迹。等到死信消费者再读到这条消息时就能结合这里的日志做业务侧补偿。6. 升级完怎么验证一张表看清限流、超时、死信是否生效6.1 验证清单与操作步骤配置改完之后不上线直接信那是自己骗自己。下面是我每次升级 RabbitMQ 消费端之后必跑的验证清单照着做一遍基本心里就有底了。验证项操作步骤预期结果限流生效停掉所有消费者用脚本向队列发送 1000 条消息再启动消费者管理后台messages_unacknowledged不会超过 消费者数 乘 prefetch且趋于稳定处理超时兜底在RabbitListener方法里Thread.sleep(5分钟)观察日志达到consumer-timeout后通道被关闭消息重新投递失败消息进死信监听方法里直接抛异常且本地重试设为 3 次观察 RabbitMQ 管理台业务队列消息数不变死信队列收到 3 条对应的失败消息TTL 过期进死信给业务队列设置 5 分钟 TTL发一条消息不去消费5 分钟后消息进入死信队列业务队列清空发送失败感知故意让消息路由到不存在的队列观察应用日志returnsCallback打印消息路由失败完整信息跑这组验证的时候最实用的命令行是rabbitmqctl list_queues name messages_ready messages_unacknowledgedmessages_ready是待在队列里等待投递的消息数messages_unacknowledged是已推给消费者但还没确认的消息数。限流有没有生效看 unack 的数量能不能被 prefetch 乘消费者数限制住一目了然。6.2 长期运维的四个原则验证通过只能说上线那天是好的。真正让这套机制长期可靠靠的是运维习惯。我踩过几轮坑之后给自己定了几条规矩。第一死信队列要每天看。死信数量从零涨到几百通常不需要恐慌但要观察增长速度。如果每天都在稳定增长说明业务侧有系统性失败不是偶发抖动。第二不要为了方便把死信队列的 TTL 设成无限大。死信本身就是异常数据保留价值随时间递减建议保留 1 到 3 天到期自动清理。第三发布端确认回调里的失败日志不能只打印不处理。哪怕只是加一个简单的计数指标也要让失败可见。第四改 prefetch 或并发参数时要压测后再上线不要凭感觉拍脑袋。同样一台 4C8G 的机器prefetch 从 100 调到 300消费吞吐可能只涨不到 10%但本地缓冲里的消息积压却翻了三倍GC 压力明显变大这不划算。用命令行再查一下未确认消息数和死信队列深度这俩指标配上消费者线程数基本就能判断整个消息链路的健康状况。结尾我在实际项目里最深的一个体会是RabbitMQ 消费端的问题从来不是单点参数能解决的。prefetch 管的是消费者拉多少consumer-timeout 管的是消费线程卡了怎么办TTL 管的是消息待多久DLX 管的是失败去哪里。四者配合起来消息才会真正走完投递-处理-成功或失败的全生命周期。如果你正在用若依做快速开发搭完业务模块后不妨抽半天时间把自带那套 RabbitMQ 集成按今天的方案升级一下。这个投入比任何业务代码都值得——因为消息中间件平时不说话一出问题就是半夜值班电话连着响。当然这种事我希望你永远不要亲自经历。
