Kafka Consumer 端必须进行消息去重吗？

2026-10-24支付小周

结论

在支付、记账、清算等不能重复产生业务副作用的场景中，通常需要在业务边界实现幂等或去重，不能只依赖 Kafka 的 transport-level exactly-once guarantees。

这并不意味着每个 Consumer 都要维护一张去重表。只要下面任一机制能把“消费消息”和“业务写入”放进同一个原子边界，就能达到同样的效果：

Consumer 使用稳定的业务事件 ID，并通过数据库唯一约束实现幂等写入。
Consumer 在同一个数据库事务中写入 inbox 或去重记录，同时完成资金变更。
目标系统提供支持 idempotency key 的原子操作。
输入和输出都在 Kafka 内部，通过 Kafka transaction 完成 consume-process-produce。
使用经过验证的事务型 Kafka Connect sink，让 offset 与目标写入获得相应的一致性保证。

如果 Consumer 会修改 Kafka 之外的数据库、支付网关或其他外部系统，而这些系统又无法与 Kafka transaction 原子提交，那么业务幂等或去重基本无法省略。

Kafka 的 exactly-once 到底保证什么

Kafka 的 exactly-once semantics 有明确的适用范围。

Idempotent producer 主要处理这样的问题：Producer 在一次运行中遇到网络错误并重试，导致同一条 record 被重复追加到 partition。Kafka 会通过 producer identity 和 sequence number 识别这类重复写入。

但它无法自动判断两个独立的业务请求是否代表同一笔付款。例如：

应用调用 send()。
Kafka 已经成功提交消息。
应用在收到确认前崩溃。
应用重启后，根据业务状态重新创建并发送消息。

第二次发送可能使用新的 Producer 状态，也可能位于新的 transaction 中。对 Kafka 来说，这是一条新的合法 record，它无法知道这条 record 与之前的记录代表同一笔业务交易。

Transactional producer 和 sendOffsetsToTransaction() 可以原子完成：

读取 Kafka 输入
        +
向 Kafka 输出 topic 写入结果
        +
提交输入 offset

如果下游 Consumer 使用：

isolation.level=read_committed

它只会读取已经提交的 transaction。这个机制适合 Kafka 内部的 consume-process-produce 链路，但不会自动处理以下情况：

链路最前端的应用重复创建业务事件；
Consumer 写入普通关系数据库；
Consumer 调用支付、短信或其他外部 API；
数据库事务无法与 Kafka transaction 原子提交。

因此，更准确的说法是：Kafka EOS 能在 Kafka transaction 管理的范围内提供 exactly-once processing，但不能让任意端到端业务流程天然只执行一次。

推荐的实现方式

1. 为业务事件分配稳定的唯一 ID

事件 ID 应在业务操作首次产生时创建，之后所有重试都沿用同一个 ID：

{
  "eventId": "01J6ZK8YF2VNRJ7Q8P4W0X9C3A",
  "paymentId": "PAY-20260901-000123",
  "eventType": "PaymentCaptured",
  "amount": "128.50",
  "currency": "CNY"
}

不要在每次发送重试时重新生成 eventId，否则 Consumer 无法判断两条消息究竟是重复投递，还是两笔不同的交易。

在支付业务中，paymentId、orderId + operationType 或其他业务唯一键，通常比 Kafka offset 更适合作为幂等键。

2. 在数据库中建立唯一约束

以 PostgreSQL 为例：

CREATE TABLE processed_events (
    consumer_name VARCHAR(100) NOT NULL,
    event_id      VARCHAR(100) NOT NULL,
    processed_at  TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (consumer_name, event_id)
);

如果同一个事件需要由多个独立的业务 Consumer 处理，唯一键应包含 Consumer 或处理器名称。“记账服务已经处理”并不表示“通知服务也已经处理”。

业务表本身也应建立符合业务语义的约束：

CREATE UNIQUE INDEX uk_ledger_payment_operation
    ON ledger_entries (payment_id, operation_type);

除了通用去重表，业务表上的唯一约束还能提供最后一道防线，而且通常更加可靠。

3. 在同一个数据库事务中完成去重和业务写入

去重标记和资金变更必须原子提交。

伪代码如下：

for (ConsumerRecord<String, PaymentEvent> record : consumer.poll(Duration.ofSeconds(1))) {
    PaymentEvent event = record.value();

    try {
        database.beginTransaction();

        int inserted = database.executeUpdate("""
            INSERT INTO processed_events (consumer_name, event_id)
            VALUES (?, ?)
            ON CONFLICT DO NOTHING
            """,
            "payment-ledger-consumer",
            event.eventId()
        );

        if (inserted == 1) {
            database.executeUpdate("""
                INSERT INTO ledger_entries
                    (payment_id, operation_type, amount, currency)
                VALUES (?, 'CAPTURE', ?, ?)
                """,
                event.paymentId(),
                event.amount(),
                event.currency()
            );
        }

        database.commit();

        TopicPartition partition =
            new TopicPartition(record.topic(), record.partition());

        consumer.commitSync(Map.of(
            partition,
            new OffsetAndMetadata(record.offset() + 1)
        ));
    } catch (Exception e) {
        database.rollback();
        throw e;
    }
}

这段流程会等数据库 transaction 成功后，再提交 Kafka offset：

如果在数据库写入前崩溃，消息会被重新消费，此时业务还没有执行。
如果数据库已经提交，但进程在提交 offset 前崩溃，消息也会被重新消费，不过唯一约束会识别出重复事件。
如果消息被重复生产，相同的业务事件 ID 仍会拦截它。
Consumer 只有在数据库提交成功后才会推进 offset。

这里保证的不是“消息绝不会重复到达”，而是“重复到达不会产生重复的业务结果”。对金融系统来说，通常真正需要的是后者。

实际代码还要处理批量拉取、rebalance、重试、死信队列和数据库连接失效等问题。如果要并行处理同一个 partition，还必须避免 offset 越过尚未完成的 record。

Producer 端应配合 Outbox Pattern

如果业务数据先写入数据库，再发送 Kafka 消息，常见做法是将业务数据和 outbox event 写入同一个数据库 transaction：

BEGIN;

UPDATE payments
SET status = 'CAPTURED'
WHERE payment_id = 'PAY-20260901-000123'
  AND status = 'PENDING';

INSERT INTO outbox_events (
    event_id,
    aggregate_id,
    event_type,
    payload
) VALUES (
    '01J6ZK8YF2VNRJ7Q8P4W0X9C3A',
    'PAY-20260901-000123',
    'PaymentCaptured',
    '{"amount":"128.50","currency":"CNY"}'
);

COMMIT;

随后由 CDC 或 outbox publisher 将记录发送到 Kafka。

Outbox Pattern 解决的是“数据库已经提交，但 Kafka 消息没有可靠产生”这一原子性问题。不过，它通常仍按 at-least-once 方式发布。publisher 或 CDC reader 崩溃后，可能再次发送同一个 outbox event。因此，event_id 必须保持稳定，下游也仍然需要具备幂等处理能力。

为什么不建议只用 Kafka offset 去重

Kafka offset 能标识某个 partition 中的一次物理投递，但不适合作为可靠的业务幂等键：

同一个业务事件被 Producer 重发后，可能获得不同的 offset。
topic 重建、数据迁移或重新生产会改变 offset。
partition 数量或分区策略可能发生变化。
一个业务事件可能通过多个 topic 传播。
补数和 replay 经常会产生新的物理位置。

offset 可以管理消费进度，也能辅助诊断，但支付去重应优先使用稳定的业务 ID。

外部 API 的处理

如果 Consumer 调用支付网关等外部服务，只有数据库去重表还不够。下面的情况仍然可能发生：

外部支付调用已经成功；
Consumer 还没来得及记录结果便崩溃；
消息被重新消费；
Consumer 再次调用外部服务。

这时需要目标 API 支持 idempotency key：

POST /payments/capture
Idempotency-Key: PAY-20260901-000123-CAPTURE

每次重试都必须使用同一个 key。目标系统应原子保存该 key 和操作结果，并在收到重复请求时返回第一次执行的结果。

如果外部系统既不参与分布式 transaction，也不支持幂等键或状态查询，那么仅靠 Kafka 无法严格保证外部副作用只发生一次。这是分布式系统本身的边界，单纯调整 Consumer 提交 offset 的顺序无法彻底解决。

注意事项

去重记录不能早于业务写入单独提交，否则可能出现“已经标记为处理，但业务实际没有完成”的情况。
去重记录不能等业务成功后，再通过另一个独立 transaction 补写，否则进程在两次写入之间崩溃时，业务仍会重复执行。
不要依赖内存缓存进行金融级去重。进程重启、扩容和缓存淘汰都会破坏这项保证。
如果需要清理去重记录，其保留时间必须覆盖 Kafka 的最大重放范围、业务补数周期和上游最大重试周期。
金额变更最好采用不可变 ledger entries，并通过业务唯一约束防止重复入账。
enable.auto.commit=true 不适合需要严格控制外部副作用的处理流程。应用应在业务提交后自行管理 offset。
Kafka transaction 中的 transactional.id、幂等 Producer 和 read_committed 仍应启用，但它们只是整体方案的一部分，不能代替业务幂等。

最终目标不应表述为“Consumer 永远只收到一次消息”，因为在包含崩溃、重试和外部系统的链路中，这通常无法实现。更实际且更容易验证的目标是：允许消息重复投递，但同一业务操作无论尝试多少次，最终都只产生一次有效的资金状态变化。