Kafka 幂等性：交付保证实践

Kafka 交付保证：生产者和消费者幂等性实践

inventory-service 正在从内存存储切换到 PostgreSQL，以支持 Outbox Pattern 和交付保证。在 docker-compose.yaml 中添加一个 postgres-inventory 容器，使用镜像 postgres:17，端口 5432:5432，以及环境变量 POSTGRES_USER=postgres、POSTGRES_PASSWORD=postgres、POSTGRES_DB=inventory_db。

在 inventory-service 的 pom.xml 中，添加以下依赖：

<dependency>
  <groupId>org.postgresql</groupId>
  <artifactId>postgresql</artifactId>
  <scope>runtime</scope>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>

在 application.properties 中，指定连接：

spring.datasource.url=jdbc:postgresql://localhost:5432/inventory_db
spring.datasource.username=postgres
spring.datasource.password=postgres

创建库存表：

CREATE TABLE inventory (
    id BIGSERIAL PRIMARY KEY,
    product_name VARCHAR(256) NOT NULL UNIQUE,
    quantity INT NOT NULL CHECK (quantity >= 0)
);

INSERT INTO inventory (product_name, quantity) VALUES
('Smartphone', 5),
('Tablet', 10),
('Desktop', 6);

Inventory 实体：

@Entity
@Table(name = "inventory")
public class Inventory {
    @Id
    @Column(name = "id")
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "product_name", nullable = false, unique = true, length = 256)
    private String productName;

    @Column(name = "quantity", nullable = false)
    private Integer quantity;

    public Inventory(String productName, Integer quantity) {
        this.productName = productName;
        this.quantity = quantity;
    }
    // getters, setters
}

带有 deductStock 的 Repository：

public interface InventoryRepository extends JpaRepository<Inventory, Long> {
    @Modifying
    @Query("UPDATE Inventory i SET i.quantity = i.quantity - :quantity WHERE i.productName = :productName AND i.quantity >= :quantity")
    int deductStock(@Param("productName") String productName, @Param("quantity") Integer quantity);
}

服务中的 KafkaListener 处理程序会检查库存可用性，并发送 InventoryReservedEvent 或 InventoryRejectedEvent。

配置 acks 并理解精确一次语义

spring.kafka.producer.acks 参数控制确认机制：0 表示至多一次，1 表示部分确认，all 表示至少一次（所有副本跟随者确认）。

分布式系统中不可能实现真正的精确一次交付，这是由于“两个将军问题”。实际实现的是精确一次语义：至少一次交付 + 生产者和消费者的幂等性。

生产者幂等性：ProducerId、Epoch 和序列号

幂等性可防止因确认丢失导致的重试时产生重复消息。通过以下配置启用：

spring.kafka.producer.properties.enable.idempotence=true

Broker 为连接分配 (ProducerId, ProducerEpoch) 对：

• ProducerId：唯一的生产者标识符
• ProducerEpoch：重启时递增，用于防范僵尸生产者

每条消息都有序列号。Broker 会拒绝 ProducerId/Epoch/序列号范围内的重复消息。

僵尸生产者场景：

• 生产者崩溃，但旧实例在新的重启后“复活”
• 旧实例继续使用较高的序列号
• 新实例从 1 开始——ProducerEpoch 可区分它们

Transactional.id 可确保重启时保留 ProducerId（Spring 会调用 initTransactions()）。否则会生成新的随机 ProducerId。

关键幂等性参数

spring.kafka.producer.acks=all
spring.kafka.producer.retries=2147483647  # >0
spring.kafka.producer.properties.max.in.flight.requests.per.connection=5  # ≤5
spring.kafka.producer.properties.enable.idempotence=true

违反这些参数会禁用幂等性。将这些配置应用到 order-service 和 inventory-service。

防范消费者双重处理的幂等性

如果消费者在提交偏移量后但业务逻辑完成前崩溃，可能会两次处理同一条消息。对于 inventory-service：

• 读取 OrderPlacedEvent
• 在 PostgreSQL 中执行 deductStock()（原子操作）
• 发送 InventoryReservedEvent
• 崩溃
• 重新读取偏移量，重复执行 deductStock()——但数量已减少，UPDATE 不会生效

通过条件 UPDATE 在业务逻辑层面确保幂等性。

该方法的优势：

• 无需额外偏移量存储
• 利用数据库 ACID 事务
• 通过 WHERE quantity >= :quantity 简单实现

关键要点

• 生产者幂等性：enable.idempotence=true + acks=all + retries>0 + max.in.flight≤5
• ProducerEpoch：重启时防范僵尸生产者
• Transactional.id：保留 ProducerId，需要 initTransactions()
• 消费者：数据库中条件 UPDATE 的业务逻辑确保幂等性
• 精确一次语义：至少一次 + 幂等性，并非字面上的精确一次

— Editorial Team