PAYATHON 2026

Spring Cloud Stream Kafka 如何实现幂等性

AI 小顾

结论

不能只凭 Kafka offset、Spring Message 的 id header，或者“以前是否见过这条消息”来决定是否丢弃。更可靠的处理方式是：

  1. 为业务命令设置稳定的幂等键，例如 orderId + commandType，或者使用独立的 commandId。
  2. 在 Product Service 的同一个本地数据库事务中完成以下操作：
    • 判断该业务操作是否已经完成；
    • 更新库存或预留状态；
    • 记录处理结果；
    • 写入待发送的响应事件。
  3. 使用唯一约束或条件更新处理并发重复消费。
  4. 通过 Transactional Outbox 可靠地发布 PRODUCT_UNRESERVATION_COMPLETED。

判断依据不应是“消息是否重复到达”，而应是“这次业务操作是否已经成功提交”。

为什么只检查消息 header 不够

消息重复不等于业务重复。

Order Service 重试时，可能会重新构造消息。新消息的 Spring MessageHeaders.ID 和 Kafka offset 都会变化，但它表达的仍然是同一个业务操作，也就是“取消订单 123 的库存预留”。

另一种情况是，Product Service 已经收到相同的 commandId，却在事务提交前崩溃。由于操作尚未完成，服务重启后仍然需要再次执行。

所以，幂等状态必须保存在 Product Service 的数据库中，并与库存变更放在同一个事务里。内存缓存、Kafka header，或者单独提交的“已处理消息表”，都无法独立消除崩溃窗口。

推荐的数据模型

与单纯维护 processed_messages 相比，为每个订单保存库存预留状态通常更稳妥：

CREATE TABLE product_reservation (
    order_id       VARCHAR(64) PRIMARY KEY,
    product_id     VARCHAR(64) NOT NULL,
    quantity       INTEGER NOT NULL,
    status         VARCHAR(20) NOT NULL,
    updated_at     TIMESTAMP NOT NULL
);

CREATE TABLE event_outbox (
    event_id       VARCHAR(64) PRIMARY KEY,
    aggregate_id  VARCHAR(64) NOT NULL,
    event_type     VARCHAR(64) NOT NULL,
    payload        TEXT NOT NULL,
    published      BOOLEAN NOT NULL DEFAULT FALSE,
    created_at     TIMESTAMP NOT NULL
);

product_reservation.status 可以使用以下状态：

RESERVED -> RELEASED

对于同一个 orderId：

  • 状态为 RESERVED 时，允许归还一次库存，随后将状态改为 RELEASED。
  • 状态为 RELEASED 时，说明库存已经归还，不能再次增加库存。
  • 如果找不到预留记录，应根据业务约定报错、延迟重试或记录异常事件，不能直接增加库存。

这个模型记录的是业务事实，比“某条 Kafka 消息是否处理过”更便于审计和修复。

消费端处理示例

下面使用 Spring Cloud Stream 函数式消费者演示。具体的绑定配置会因 Spring Cloud Stream 版本而异，但事务边界的处理原则相同。

public record UndoProductReservation(
        String commandId,
        String orderId,
        String productId,
        int quantity
) {
}
@Configuration
public class ProductConsumers {

    @Bean
    public Consumer<Message<UndoProductReservation>> undoProductReservation(
            ProductReservationService service) {

        return message -> service.undo(message.getPayload());
    }
}

核心逻辑放在一个数据库事务中：

@Service
public class ProductReservationService {

    private final ReservationRepository reservationRepository;
    private final ProductRepository productRepository;
    private final OutboxRepository outboxRepository;

    public ProductReservationService(
            ReservationRepository reservationRepository,
            ProductRepository productRepository,
            OutboxRepository outboxRepository) {
        this.reservationRepository = reservationRepository;
        this.productRepository = productRepository;
        this.outboxRepository = outboxRepository;
    }

    @Transactional
    public void undo(UndoProductReservation command) {
        ProductReservation reservation =
                reservationRepository.findByOrderIdForUpdate(command.orderId())
                        .orElseThrow(() ->
                                new IllegalStateException(
                                        "Reservation not found: " + command.orderId()));

        if (reservation.getStatus() == ReservationStatus.RELEASED) {
            ensureCompletionEvent(command);
            return;
        }

        if (reservation.getStatus() != ReservationStatus.RESERVED) {
            throw new IllegalStateException(
                    "Cannot release reservation in state: "
                            + reservation.getStatus());
        }

        int updated = productRepository.increaseStock(
                reservation.getProductId(),
                reservation.getQuantity());

        if (updated != 1) {
            throw new IllegalStateException(
                    "Stock update failed: " + reservation.getProductId());
        }

        reservation.markReleased();
        reservationRepository.save(reservation);

        ensureCompletionEvent(command);
    }

    private void ensureCompletionEvent(UndoProductReservation command) {
        outboxRepository.insertIfAbsent(
                "product-unreservation-completed:" + command.orderId(),
                command.orderId(),
                "PRODUCT_UNRESERVATION_COMPLETED",
                createPayload(command));
    }

    private String createPayload(UndoProductReservation command) {
        return """
                {"orderId":"%s","productId":"%s"}
                """.formatted(command.orderId(), command.productId());
    }
}

findByOrderIdForUpdate 可以通过悲观锁实现：

public interface ReservationRepository
        extends JpaRepository<ProductReservation, String> {

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("""
           select r
           from ProductReservation r
           where r.orderId = :orderId
           """)
    Optional<ProductReservation> findByOrderIdForUpdate(String orderId);
}

event_outbox.event_id 必须有唯一约束。insertIfAbsent 可以使用数据库提供的冲突忽略语法，也可以先尝试插入，再捕获唯一键冲突。具体 SQL 取决于使用的数据库。

崩溃时会发生什么

这个事务设计可以覆盖几个主要的崩溃窗口。

更新库存之前崩溃

事务没有提交，库存和预留状态都不会发生变化。消息重试后仍会读到 RESERVED，系统将重新执行库存归还。

更新库存之后、事务提交之前崩溃

整个数据库事务都会回滚，其中包括：

  • 增加库存；
  • 将状态改为 RELEASED；
  • 写入 Outbox 事件。

消息重试后可以安全地再次执行。

数据库事务提交之后、响应发布之前崩溃

此时库存已经增加，状态已经变为 RELEASED，Outbox 中也已经写入响应事件。

消息再次到达时，系统不会重复增加库存。独立的 Outbox 发布器会继续发送 PRODUCT_UNRESERVATION_COMPLETED。

响应已经发送，但 Outbox 尚未标记完成时崩溃

Outbox 发布器可能重复发送响应。这是至少一次投递的正常情况，因此 Order Service 也需要使用 eventId 或业务状态机对响应做幂等处理。

Outbox 发布

后台任务负责读取尚未发布的事件，并将它们发送到 Kafka：

@Component
public class OutboxPublisher {

    private final OutboxService outboxService;

    public OutboxPublisher(OutboxService outboxService) {
        this.outboxService = outboxService;
    }

    @Scheduled(fixedDelayString = "${outbox.publish-delay-ms:1000}")
    public void publish() {
        outboxService.publishPendingEvents();
    }
}

实现时，不能简单地在同一个长事务中先发送 Kafka 消息，再把 published 改为 true，因为数据库和 Kafka 之间仍有故障窗口。常见做法是接受 Outbox 至少发布一次，并要求接收方实现幂等处理。也可以使用 CDC 工具，将 Outbox 表的变更转发到 Kafka。

如果只使用已处理消息表

也可以创建如下表结构：

CREATE TABLE processed_command (
    consumer_name VARCHAR(100) NOT NULL,
    command_id    VARCHAR(64) NOT NULL,
    processed_at  TIMESTAMP NOT NULL,
    PRIMARY KEY (consumer_name, command_id)
);

不过，插入 processed_command、更新库存和写入 Outbox 必须放在同一个事务中：

@Transactional
public void handle(UndoProductReservation command) {
    boolean firstProcessing =
            processedCommandRepository.insertIfAbsent(
                    "product-service",
                    command.commandId());

    if (!firstProcessing) {
        return;
    }

    productRepository.increaseStock(
            command.productId(),
            command.quantity());

    outboxRepository.insert(...);
}

如果先提交“已处理”记录，再更新库存，中途崩溃会导致后续重试被错误丢弃。如果先提交库存变更，再写入处理记录，中途崩溃又会导致库存重复增加。因此，这些操作必须原子提交。

在库存预留场景中，业务状态表通常比通用的 processed_command 表更合适。它不仅能防止重复处理，还能检查订单是否确实预留过库存、释放数量是否正确，以及当前状态是否合法。

其他注意事项

  • Kafka producer 的幂等发送只能减少部分生产端的重复写入，不能代替业务幂等。
  • Kafka exactly-once semantics 主要处理从 Kafka 读取、处理并写回 Kafka 的事务链路，无法自动实现普通关系数据库与 Kafka 的原子提交。
  • 同一订单的事件可以使用 orderId 作为 Kafka key，让它们进入同一分区并保持顺序。即使这样，数据库端仍然需要使用唯一约束或锁来处理并发。
  • 消费失败时应抛出异常，让消息进入重试流程。不要捕获异常后正常返回，否则框架可能会提交 offset。
  • 重试次数超过限制后，应将消息送入 DLQ，并保留 commandId、orderId、异常原因和原始 payload，以便人工补偿。
  • UNDO_PRODUCT_RESERVATION 最好携带稳定的 commandId，但最终约束仍应落在 orderId 对应的预留状态上，防止协调者错误地生成多个不同的 commandId。

核心原则是：消息可以重复投递，业务状态不能重复迁移。只要库存变更、状态迁移和 Outbox 写入处于同一个数据库事务中，Product Service 就能准确判断上一次处理究竟是尚未完成、需要重试，还是已经完成、只需返回已有结果。

备注：内容仅供参考。