Hello MQ

外观

幂等消费 ​

本页结论：at-least-once 投递下重复不是意外而是常态，业务必须预期重复。幂等消费的标准做法是：processed_messages 唯一键与业务写入放在同一个本地事务里，事务提交之后才 ACK/提交 offset。「业务已提交、确认未发出」的崩溃窗口无法消除，只能靠幂等表兜住。

为什么重复不可避免 ​

四个产品的确认机制不同（RabbitMQ ACK、Kafka offset 提交、RocketMQ 消费位点、Pulsar ack），但失败窗口的位置相同：业务副作用已生效，向 Broker 的确认还没发出。此时 Broker 认为消息未确认，必然重投。反过来若先确认再写业务，窗口就变成丢消息。顺序只能二选一，本仓库的选择是：业务事务提交 → 确认，重复交给幂等表处理（完整推导见投递语义）。

§5.4 基准实现（所有产品通用） ​

text
1. 开启本地数据库事务
2. 尝试将 messageId 插入 processed_messages（唯一键）
3. 唯一键冲突 → 记录 duplicate_skipped，安全确认消息
4. 首次处理 → 执行业务写入，提交本地事务
5. 数据库提交成功后才 ACK / commitSync / 提交消费位点

两个细节决定正确性：

  • 第 2 步与第 4 步必须同事务。如果幂等键先单独提交、业务写入后失败，重投到来时会被误判为「已处理」而跳过——业务永远不会执行。
  • 第 5 步永远在最后。确认是「业务已持久化」的通知，不是业务本身的一部分。

崩溃窗口：已验证的实验证据 ​

本仓库用崩溃注入复现了这个窗口（第 2 步成功、第 5 步前 Runtime.halt(137)，重启后消息带 redelivered=true 到达，被幂等表拦截为 duplicate_skipped，业务表恰好 3 行）：

rabbitmq / consumer-crashbash demos/rabbitmq/consumer-crash/run.sh
assert.out.txt
PASS crashExitCode: 137
PASS crashAfterBusinessCommit: 1
PASS receivedTotal: 4
PASS redeliveredCount: 1
PASS duplicatesObserved: 1
PASS uniqueMessageIds: 3
PASS business_rows: 3
PASS queueDepthAfter: 0
PASS consumer-run2.exitCode: 0
RESULT: all assertions passed
consumer-run1.out.txt
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
[consumer] timestamp=2026-10-07T10:27:14.314008641Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.crash consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:27:14.322479817Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=4fac7be9f8ad-1-after-delivery-01JMQ000000000000000000001-1 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:15.327335311Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=4fac7be9f8ad-1-after-delivery-01JMQ000000000000000000001-1 status=released
[consumer] timestamp=2026-10-07T10:27:15.337218764Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.crash attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:27:15.337699711Z checkpoint=before-ack messageId=01JMQ000000000000000000001 token=4fac7be9f8ad-1-before-ack-01JMQ000000000000000000001-2 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:16.340446936Z checkpoint=before-ack messageId=01JMQ000000000000000000001 token=4fac7be9f8ad-1-before-ack-01JMQ000000000000000000001-2 status=released
[consumer] timestamp=2026-10-07T10:27:16.340660034Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.crash status=crash_injected
consumer-run2.out.txt
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
[consumer] timestamp=2026-10-07T10:27:18.331903782Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.crash consumer=consumer-1 attempt=1 redelivered=true status=received
[replay-gate] timestamp=2026-10-07T10:27:18.340119726Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=2cbfb1260f3d-1-after-delivery-01JMQ000000000000000000001-1 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:19.244860771Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=2cbfb1260f3d-1-after-delivery-01JMQ000000000000000000001-1 status=released
[consumer] timestamp=2026-10-07T10:27:19.251586332Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.crash attempt=1 status=duplicate_skipped
[replay-gate] timestamp=2026-10-07T10:27:19.251975513Z checkpoint=before-ack messageId=01JMQ000000000000000000001 token=2cbfb1260f3d-1-before-ack-01JMQ000000000000000000001-2 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:20.253760487Z checkpoint=before-ack messageId=01JMQ000000000000000000001 token=2cbfb1260f3d-1-before-ack-01JMQ000000000000000000001-2 status=released
[consumer] timestamp=2026-10-07T10:27:20.255434279Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.crash consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:27:20.255920404Z checkpoint=after-delivery messageId=01JMQ000000000000000000002 token=2cbfb1260f3d-1-after-delivery-01JMQ000000000000000000002-3 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:21.257916980Z checkpoint=after-delivery messageId=01JMQ000000000000000000002 token=2cbfb1260f3d-1-after-delivery-01JMQ000000000000000000002-3 status=released
[consumer] timestamp=2026-10-07T10:27:21.263174006Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.crash attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:27:21.263516787Z checkpoint=before-ack messageId=01JMQ000000000000000000002 token=2cbfb1260f3d-1-before-ack-01JMQ000000000000000000002-4 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:22.165093163Z checkpoint=before-ack messageId=01JMQ000000000000000000002 token=2cbfb1260f3d-1-before-ack-01JMQ000000000000000000002-4 status=released
[consumer] timestamp=2026-10-07T10:27:22.166557146Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.crash consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:27:22.166974929Z checkpoint=after-delivery messageId=01JMQ000000000000000000003 token=2cbfb1260f3d-1-after-delivery-01JMQ000000000000000000003-5 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:23.168666227Z checkpoint=after-delivery messageId=01JMQ000000000000000000003 token=2cbfb1260f3d-1-after-delivery-01JMQ000000000000000000003-5 status=released
[consumer] timestamp=2026-10-07T10:27:23.170617117Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.crash attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:27:23.171131674Z checkpoint=before-ack messageId=01JMQ000000000000000000003 token=2cbfb1260f3d-1-before-ack-01JMQ000000000000000000003-6 status=waiting
[replay-gate] timestamp=2026-10-07T10:27:24.273522547Z checkpoint=before-ack messageId=01JMQ000000000000000000003 token=2cbfb1260f3d-1-before-ack-01JMQ000000000000000000003-6 status=released
[consumer] timestamp=2026-10-07T10:27:24.274089233Z level=INFO service=order-service product=rabbitmq lab=consumer-crash queue=orders.crash received=3 status=done
inspect-db.out.txt
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
[inspect] timestamp=2026-10-07T10:27:24.716719592Z level=INFO service=order-service product=rabbitmq lab=consumer-crash business_rows=3 processed_rows=3 status=snapshot
producer.out.txt
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
[producer] timestamp=2026-10-07T10:27:13.597266852Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.crash routingKey=orders.crash durationMs=2 status=confirmed
[producer] timestamp=2026-10-07T10:27:13.599968622Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.crash routingKey=orders.crash durationMs=0 status=confirmed
[producer] timestamp=2026-10-07T10:27:13.600894393Z level=INFO service=order-service product=rabbitmq lab=consumer-crash messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.crash routingKey=orders.crash durationMs=0 status=confirmed
[producer] timestamp=2026-10-07T10:27:13.600975954Z level=INFO service=order-service product=rabbitmq lab=consumer-crash destination=orders.crash confirmed=3 status=done
setup.out.txt
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

完整实验步骤与断言解读见 RabbitMQ 可靠性。Kafka 分卷从 offset 语义给出同样的结论（见 Kafka 可靠性「崩溃窗口与幂等消费」）：「提交 offset 等于业务已成功」是禁止表述——那是两个系统上的两个独立动作。

去重键怎么选 ​

选择适用注意
messageId（默认）通用；信封必填，天然全局唯一生产者重发若生成新 messageId，去重失效——Outbox 要求 messageId 在写发件箱时就固定
业务键（如 orderId + 操作类型）生产者可能用不同 messageId 重发同一业务动作键要能代表「同一个业务意图」，不是「同一条消息」
天然幂等的业务写法状态机式写入（UPDATE … SET status='PAID' WHERE status='UNPAID'）幂等表仍然建议保留，用于观测与防回归

其他工程点：

  • 幂等记录要有保留策略（按业务窗口保留，如 7～30 天后归档），否则表无限增长；归档窗口外的极老重投按新消息处理，业务键约束兜底。
  • 用 Redis SETNX 做去重时，注意它与业务写入不在同一事务：标记成功、业务失败后 Redis 键还在，重投会被错误跳过。分布式锁/缓存只适合做前置快筛，最终裁决要落在与业务同库的唯一键上。
  • 不要根据 redelivered=true 直接跳过：重投的消息可能第一次就没处理完，必须走完整幂等流程（见 RabbitMQ 可靠性）。

与重试的关系 ​

重试会放大重复：同一条消息每次重试都可能在「业务已提交、未确认」处失败。因此重试与 DLQ的消费逻辑必须复用同一套幂等流程，而不是在重试路径上绕过它。观测上，duplicatesObserved/重复拦截数应与重投率一起看（见可观测性统一指标）。

保证成立的条件 / 不保证什么 ​

  • 条件：messageId 全局唯一且生产者不随意更换；幂等表与业务同库同事务；确认动作在事务之后。
  • 不保证：业务副作用「恰好一次」中外部系统部分——调用短信、第三方 HTTP 这类无法回读校验的副作用，需要在调用前先落幂等标记并容忍「可能已发出」。
  • 不消除崩溃窗口本身，只是让窗口的后果从「重复落库」变成「一次 duplicate_skipped 日志」。

常见误区 ​

  • 「重复是小概率，可以不管」——at-least-once 的含义是业务必须预期重复，不是「偶尔可能重复」。
  • 「先 ACK 再写库更安全」——方向反了，那会把重复窗口换成丢失窗口。
  • 「幂等表只是性能优化」——它是崩溃窗口下的正确性组件，不可省略。
  • 「Kafka 开了事务/EOS 就不用幂等了」——EOS 只覆盖 Kafka 内部 topic→topic，外部数据库副作用仍需幂等（见 Kafka 可靠性）。

官方资料 ​

以统一实验验证消息系统语义边界

MIT License