Hello MQ

外观

Apache Kafka 可靠性 ​

本页结论：Kafka 的可靠性由三处配置共同决定——生产端 acks + 幂等/事务、Broker 端副本与 ISR、消费端 offset 提交时机。「提交 offset」与「业务提交」是两个独立动作，之间的崩溃窗口靠幂等消费兜底；事务只覆盖 Kafka 内部，不等于外部系统 exactly-once。

生产端：acks、重试与幂等 ​

acks确认条件丢失窗口
0不等确认网络/Broker 故障即丢
1Leader 写入内存/页缓存Leader 崩溃且未同步到 Follower → 丢
all（-1）ISR 全部副本写入取决于 ISR 收缩与 min.insync.replicas

配套关系：

  • acks=all 必须配合副本数 ≥2 才有意义；ISR 只剩 Leader 时 all 退化为 1。min.insync.replicas 让 Broker 在 ISR 不足时拒绝写入（宁可不可用，不假确认）。
  • Producer 重试：网络抖动时客户端自动重发，非幂等配置下可能产生重复记录（同内容两个 offset）。
  • Idempotent Producer（enable.idempotence=true，本仓库默认）：为每条记录带 producer id + 序列号，Broker 去重，重试不再产生重复。它是事务的前置条件。
java
props.put("acks", "all");
props.put("enable.idempotence", "true");
RecordMetadata md = producer.send(record).get(); // partition + offset

消费端：offset 提交即「确认」 ​

Kafka 没有单条 ACK；消费进度就是已提交 offset。

模式行为风险
自动提交（enable.auto.commit=true）按时间间隔提交「已 poll 到」的位点处理中崩溃 → 已提交但未处理完 = 丢；提交后重新 poll 前崩溃边界附近 = 重复
手动提交（commitSync/commitAsync）业务决定提交时机提交早于业务完成同样有丢窗口

崩溃窗口与幂等消费（§5.4 基准实现） ​

正确顺序：业务事务提交 → commitSync。

text
1. 开启本地数据库事务
2. 插入 messageId 到 processed_messages（唯一键）
3. 唯一键冲突 → duplicate_skipped
4. 首次处理 → 执行业务写入，提交事务
5. 提交成功后才 commitSync（含该分区新位点）

第 4 步成功、第 5 步前崩溃 → 重启后从旧 offset 重读 → 幂等表拦截。该窗口不可消除。「提交 offset 等于业务数据库已成功提交」是错误表述——两者是独立系统上的两个动作。本仓库 kafka basic 实验即按此实现：

kafka / basicbash demos/kafka/basic/run.sh
assert.out.txt
PASS produced: 3
PASS received: 3
PASS uniqueMessageIds: 3
PASS businessCommitted: 3
PASS business_rows: 3
PASS consumerGroupLag: 0
PASS consumer.exitCode: 0
RESULT: all assertions passed
consumer.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:29:46.771963444Z level=INFO service=order-service product=kafka lab=basic destination=orders.basic consumerGroup=orders-basic-group consumer=consumer-1 partitions=0,1,2 status=assigned
[consumer] timestamp=2026-10-07T10:29:46.964422440Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.basic partitionOrQueue=0 offset=0 seq=1 consumerGroup=orders-basic-group consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:29:46.970174007Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=a335bb90533f-1-after-delivery-01JMQ000000000000000000001-1 status=waiting
[replay-gate] timestamp=2026-10-07T10:29:51.079078733Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=a335bb90533f-1-after-delivery-01JMQ000000000000000000001-1 status=released
[consumer] timestamp=2026-10-07T10:29:51.100475371Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.basic partitionOrQueue=0 seq=1 consumerGroup=orders-basic-group attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:29:51.100938836Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000001 token=a335bb90533f-1-before-offset-commit-01JMQ000000000000000000001-2 status=waiting
[replay-gate] timestamp=2026-10-07T10:29:54.708154139Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000001 token=a335bb90533f-1-before-offset-commit-01JMQ000000000000000000001-2 status=released
[consumer] timestamp=2026-10-07T10:29:54.725218269Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.basic partitionOrQueue=0 offset=1 seq=2 consumerGroup=orders-basic-group consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:29:54.725755269Z checkpoint=after-delivery messageId=01JMQ000000000000000000002 token=a335bb90533f-1-after-delivery-01JMQ000000000000000000002-3 status=waiting
[replay-gate] timestamp=2026-10-07T10:29:58.932308515Z checkpoint=after-delivery messageId=01JMQ000000000000000000002 token=a335bb90533f-1-after-delivery-01JMQ000000000000000000002-3 status=released
[consumer] timestamp=2026-10-07T10:29:58.935799896Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.basic partitionOrQueue=0 seq=2 consumerGroup=orders-basic-group attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:29:58.936243885Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000002 token=a335bb90533f-1-before-offset-commit-01JMQ000000000000000000002-4 status=waiting
[replay-gate] timestamp=2026-10-07T10:30:02.742259978Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000002 token=a335bb90533f-1-before-offset-commit-01JMQ000000000000000000002-4 status=released
[consumer] timestamp=2026-10-07T10:30:02.749165574Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.basic partitionOrQueue=1 offset=0 seq=3 consumerGroup=orders-basic-group consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:30:02.749598003Z checkpoint=after-delivery messageId=01JMQ000000000000000000003 token=a335bb90533f-1-after-delivery-01JMQ000000000000000000003-5 status=waiting
[replay-gate] timestamp=2026-10-07T10:30:06.655833646Z checkpoint=after-delivery messageId=01JMQ000000000000000000003 token=a335bb90533f-1-after-delivery-01JMQ000000000000000000003-5 status=released
[consumer] timestamp=2026-10-07T10:30:06.657898866Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.basic partitionOrQueue=1 seq=3 consumerGroup=orders-basic-group attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:30:06.658327900Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000003 token=a335bb90533f-1-before-offset-commit-01JMQ000000000000000000003-6 status=waiting
[replay-gate] timestamp=2026-10-07T10:30:10.865217961Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000003 token=a335bb90533f-1-before-offset-commit-01JMQ000000000000000000003-6 status=released
[consumer] timestamp=2026-10-07T10:30:10.868286305Z level=INFO service=order-service product=kafka lab=basic destination=orders.basic consumer=consumer-1 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:30:11.499099004Z level=INFO service=order-service product=kafka lab=basic 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:29:45.278255208Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.basic partitionOrQueue=0 offset=0 seq=1 durationMs=261 status=produced
[producer] timestamp=2026-10-07T10:29:45.289933444Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.basic partitionOrQueue=0 offset=1 seq=2 durationMs=8 status=produced
[producer] timestamp=2026-10-07T10:29:45.298217877Z level=INFO service=order-service product=kafka lab=basic messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.basic partitionOrQueue=1 offset=0 seq=3 durationMs=7 status=produced
[producer] timestamp=2026-10-07T10:29:45.298316623Z level=INFO service=order-service product=kafka lab=basic destination=orders.basic produced=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.

事务的边界 ​

Transactional Producer 把一个事务内的多条 produce（可加 sendOffsetsToTransaction 把消费位点一起原子提交）变成原子可见：

  • read_committed 消费者只能看到已 commitTransaction 的消息；abortTransaction 的消息永不可见。
  • 幂等 producer 保证序列号无重复；事务协调器（transaction coordinator）管理事务状态。

动手验证（3 条 commit 事务可见，2 条 abort 事务不可见）：

bash
bash demos/kafka/idempotence-transaction/run.sh
kafka / idempotence-transactionbash demos/kafka/idempotence-transaction/run.sh
assert.out.txt
PASS txnCommitted: 1
PASS txnAborted: 1
PASS committedVisible: 3
PASS abortedVisible: 0
PASS business_rows: 3
PASS rawLagIncludesAbortedControlRecord: 1
PASS consumer.exitCode: 0
RESULT: all assertions passed
consumer.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:32:24.486536623Z level=INFO service=order-service product=kafka lab=idempotence-transaction destination=orders.txn consumerGroup=orders-txn-group consumer=consumer-1 partitions=0,1,2 status=assigned
[consumer] timestamp=2026-10-07T10:32:24.765245543Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.txn partitionOrQueue=0 offset=0 seq=1 consumerGroup=orders-txn-group consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:32:24.773017977Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=a73392743e88-1-after-delivery-01JMQ000000000000000000001-1 status=waiting
[replay-gate] timestamp=2026-10-07T10:32:28.881668840Z checkpoint=after-delivery messageId=01JMQ000000000000000000001 token=a73392743e88-1-after-delivery-01JMQ000000000000000000001-1 status=released
[consumer] timestamp=2026-10-07T10:32:28.891448512Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.txn partitionOrQueue=0 seq=1 consumerGroup=orders-txn-group attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:32:28.892000097Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000001 token=a73392743e88-1-before-offset-commit-01JMQ000000000000000000001-2 status=waiting
[replay-gate] timestamp=2026-10-07T10:32:33.199286348Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000001 token=a73392743e88-1-before-offset-commit-01JMQ000000000000000000001-2 status=released
[consumer] timestamp=2026-10-07T10:32:33.215472751Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.txn partitionOrQueue=0 offset=1 seq=2 consumerGroup=orders-txn-group consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:32:33.215895160Z checkpoint=after-delivery messageId=01JMQ000000000000000000002 token=a73392743e88-1-after-delivery-01JMQ000000000000000000002-3 status=waiting
[replay-gate] timestamp=2026-10-07T10:32:37.023578222Z checkpoint=after-delivery messageId=01JMQ000000000000000000002 token=a73392743e88-1-after-delivery-01JMQ000000000000000000002-3 status=released
[consumer] timestamp=2026-10-07T10:32:37.026444845Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.txn partitionOrQueue=0 seq=2 consumerGroup=orders-txn-group attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:32:37.026845586Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000002 token=a73392743e88-1-before-offset-commit-01JMQ000000000000000000002-4 status=waiting
[replay-gate] timestamp=2026-10-07T10:32:41.237735455Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000002 token=a73392743e88-1-before-offset-commit-01JMQ000000000000000000002-4 status=released
[consumer] timestamp=2026-10-07T10:32:41.245433658Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.txn partitionOrQueue=1 offset=0 seq=3 consumerGroup=orders-txn-group consumer=consumer-1 attempt=1 redelivered=false status=received
[replay-gate] timestamp=2026-10-07T10:32:41.245932831Z checkpoint=after-delivery messageId=01JMQ000000000000000000003 token=a73392743e88-1-after-delivery-01JMQ000000000000000000003-5 status=waiting
[replay-gate] timestamp=2026-10-07T10:32:45.052945832Z checkpoint=after-delivery messageId=01JMQ000000000000000000003 token=a73392743e88-1-after-delivery-01JMQ000000000000000000003-5 status=released
[consumer] timestamp=2026-10-07T10:32:45.055556453Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.txn partitionOrQueue=1 seq=3 consumerGroup=orders-txn-group attempt=1 status=business_committed
[replay-gate] timestamp=2026-10-07T10:32:45.055976925Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000003 token=a73392743e88-1-before-offset-commit-01JMQ000000000000000000003-6 status=waiting
[replay-gate] timestamp=2026-10-07T10:32:49.363215906Z checkpoint=before-offset-commit messageId=01JMQ000000000000000000003 token=a73392743e88-1-before-offset-commit-01JMQ000000000000000000003-6 status=released
[consumer] timestamp=2026-10-07T10:32:49.366423565Z level=INFO service=order-service product=kafka lab=idempotence-transaction destination=orders.txn consumer=consumer-1 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:32:50.025508354Z level=INFO service=order-service product=kafka lab=idempotence-transaction business_rows=3 processed_rows=3 status=snapshot
producer-abort.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:32:22.893161119Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.txn partitionOrQueue=0 offset=3 seq=1 durationMs=41 status=produced
[producer] timestamp=2026-10-07T10:32:22.904514623Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.txn partitionOrQueue=0 offset=4 seq=2 durationMs=8 status=produced
[producer] timestamp=2026-10-07T10:32:22.908312981Z level=INFO service=order-service product=kafka lab=idempotence-transaction destination=orders.txn produced=2 status=txn_aborted
producer-commit.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:32:21.713775686Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000001 eventType=order.created schemaVersion=1 aggregateId=order-1001 traceId=00000000000000000000000000000001 correlationId=order-1001 destination=orders.txn partitionOrQueue=0 offset=0 seq=1 durationMs=62 status=produced
[producer] timestamp=2026-10-07T10:32:21.727573215Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000002 eventType=order.created schemaVersion=1 aggregateId=order-1002 traceId=00000000000000000000000000000002 correlationId=order-1002 destination=orders.txn partitionOrQueue=0 offset=1 seq=2 durationMs=10 status=produced
[producer] timestamp=2026-10-07T10:32:21.739508555Z level=INFO service=order-service product=kafka lab=idempotence-transaction messageId=01JMQ000000000000000000003 eventType=order.created schemaVersion=1 aggregateId=order-1003 traceId=00000000000000000000000000000003 correlationId=order-1003 destination=orders.txn partitionOrQueue=1 offset=0 seq=3 durationMs=11 status=produced
[producer] timestamp=2026-10-07T10:32:21.754937670Z level=INFO service=order-service product=kafka lab=idempotence-transaction destination=orders.txn produced=3 status=txn_committed
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.

边界必须说清楚（禁止表述之三）：

  • Kafka EOS 的 exactly-once 指 Kafka 内部（topic → topic 的 produce-consume）不重不丢；
  • 开启事务后，任意外部系统副作用（数据库、HTTP、短信）都不是 exactly-once——外部写入仍需幂等设计。

顺序与重试的关系 ​

  • 分区内顺序由写入顺序决定；Producer 开 max.in.flight.requests.per.connection>1 且非幂等时，重试可能乱序。幂等 producer 允许最多 5 个在途请求仍保序。
  • 消费端「失败重试」通常意味着跳过或转发该消息（应用层 DLQ），没有 RabbitMQ 式 requeue 到队头。

三层语义总结 ​

层级保证条件
Broker确认的消息在 ISR 全部副本可读acks=all + min.insync.replicas≥2（多副本）；本仓库单节点 RF=1 仅覆盖 Leader 存活场景
Client不重发丢、不提前提交幂等 producer + 异常重试；手动 commitSync 且晚于业务提交
Business效果恰好一次幂等表 + 本地事务；外部系统副作用不在 Kafka 事务覆盖内

常见误区 ​

  • 「acks=all 就绝对不丢」——ISR 收缩到 1 时确认不再要求第二个副本；必须与 min.insync.replicas 一起配置。
  • 「事务提交后所有下游都恰好一次」——见上文边界说明。
  • 「消费者崩溃只重投一条」——重投单位是「已提交 offset 之后的全部消息」。

官方资料 ​

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

MIT License