Spring AMQP:监听容器并发、预取与确认模式

并发

实测 · 本机执行输出

  [默认(concurrency=1)] 队列上的消费者数 1;20 条处理完共 4.1 秒;参与的线程 ['#0-1']
  [concurrency=4] 队列上的消费者数 4;20 条处理完共 1.0 秒;参与的线程 ['#0-1', '#0-2', '#0-3', '#0-4']
  [concurrency=1, max-concurrency=4] 队列上的消费者数 1;20 条处理完共 4.1 秒;参与的线程 ['#0-1']

concurrency 启动时创建;max-concurrency 对短突发不起作用。

预取

实测 · 本机执行输出

  [prefetch 默认(250)] 刚发完 0.4 秒时队列 ready=0、unacked=5;10 条全部处理完共 5.0 秒;各线程处理条数 {'#0-1': 5, '#0-2': 5},其中 1000 ms 的慢消息 {'#0-2': 5}
  [prefetch=1] 刚发完 0.4 秒时队列 ready=7、unacked=2;10 条全部处理完共 3.1 秒;各线程处理条数 {'#0-1': 5, '#0-2': 5},其中 1000 ms 的慢消息 {'#0-1': 3, '#0-2': 2}

默认 250 让慢消息集中在一个消费者上。

异常

实测 · 本机执行输出

  [默认(AUTO 确认,default-requeue-rejected 默认 true)] 3 秒内 poison-1 被投递 37670 次;ok-2 是否被处理 True;队列剩余 ready=0 unacked=1
  [default-requeue-rejected=false] 3 秒内 poison-1 被投递 1 次;ok-2 是否被处理 True;队列剩余 ready=0 unacked=0

默认拒绝并重新入队,形成死循环。

确认模式

实测 · 本机执行输出

  [AUTO] 处理中(消息发出 0.8 秒后):ready=0 unacked=4 consumers=1
    kill -9 之后:ready=4 unacked=0
    重启应用后收到的消息:['m2(redelivered)', 'm3(redelivered)', 'm4(redelivered)', 'm5(redelivered)'];其中标记为 redelivered 的 4 条
  [NONE(自动确认,autoAck)] 处理中(消息发出 0.8 秒后):ready=0 unacked=0 consumers=1
    kill -9 之后:ready=0 unacked=0
    重启应用后收到的消息:[];其中标记为 redelivered 的 0 条
  [MANUAL(代码里自己 basicAck;noack-1 故意不确认)] 处理中(消息发出 0.8 秒后):ready=0 unacked=5 consumers=1
    kill -9 之后:ready=5 unacked=0
    重启应用后收到的消息:['m2(redelivered)', 'm3(redelivered)', 'm4(redelivered)', 'm5(redelivered)', 'noack-1(redelivered)'];其中标记为 redelivered 的 5 条

AUTO / MANUAL 进程被杀不丢消息;NONE 会丢。

分步演示

AMQP 监听容器

本步的对象

本文用到的工具与边界

本机 macOS 27 arm64 / OpenJDK 21.0.12(Homebrew) / Spring Boot 4.0.8(Spring Framework 7.0.9、Tomcat 11.0.24) / Spring Cloud 2025.1.3;用 javac 编译、java 直接运行,依赖由 Maven 从 Maven Central 解析;数字每次运行会有小幅抖动。RabbitMQ 是 Homebrew 装的、自己起的独立实例(队列用 durable,因为新版 RabbitMQ 默认禁止非独占的瞬时队列);队列状态用 rabbitmqctl list_queues 读取(管理 API 的统计有几秒延迟,不用它);消息由 Python pika 发送;耗时用 Thread.sleep 模拟
没有验证:max-concurrency 在持续积压时的实际扩容速度与参数(consecutive-active-trigger、start-consumer-min-interval);prefetch 过小对吞吐的影响;批量监听(batch listener);消费者标签与独占消费;发布确认与持久化;重试与死信交换机(见第 139 篇);事务性通道;messages_unacknowledged 长期不降的告警

下一篇 · SPRING 139

Spring AMQP 的重试与死信