使用 Kafka 的 Outbox 模式：数据库写入时不丢失事件

1. 概览

一个 service 通常需要连续执行两个操作：向数据库保存数据并向 Kafka 发送事件给其他 service。这两个系统没有共同的 transaction。如果一步成功而另一步失败，系统将失去一致性。结果是订单已保存，但没有人收到 OrderPlaced 事件。

本文解释 dual write 问题并实现 Outbox 模式，用 Java 21、Spring Boot 3.x 和 Spring Kafka。读完本文，你将拥有一个 message relay 将事件从 outbox 表推送到 Kafka，包括测试和幂等 consumer。

2. Dual write 问题

看看下面这种“天真”的写法：

@Transactional
public Order placeOrder(String customerId, BigDecimal amount) {
    Order order = orderRepository.save(new Order(customerId, amount));
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
    return order;
}

这段代码至少有两个失败的场景：

Kafka 发送成功，transaction 回滚。 例如 commit 时出错或违反约束。幽灵事件已经发出，对应一个不存在的订单。
Transaction commit，Kafka 发送失败。 Broker 可能不可达，或者应用在 commit 后立即崩溃。订单在数据库中但事件消失了。

交换这两个命令的顺序也不能从根本上解决问题。两个操作在不同的系统上，如果没有分布式事务就不能是原子的。大多数团队不想使用 XA/2PC，因为它复杂且缓慢。

3. Outbox 模式的思想

Outbox 模式将“写数据库 + 发送 Kafka”变成“写数据库 + 写数据库”：

在同一个 transaction 中，service 保存 Order 并向 outbox_event 表插入一行。
一个叫 message relay 的单独进程读取 outbox 表中未发送的行，将它们发送到 Kafka，然后标记为已发送。

因为第 1 步只涉及一个数据库，它是完全原子的：有订单就一定有事件。第 2 步可以重试任意次数。代价是你得到 at-least-once 语义：一个事件可能被发送多次，但永远不会丢失。

4. 用 Spring Boot 实现

4.1. 依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

接下来是安全的 producer 配置：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.properties.enable.idempotence=true

acks=all 强制 broker 在所有副本同步写入后确认。enable.idempotence=true 帮助 producer 在网络层重试时不自动生成重复。

4.2. Outbox 表和 entity

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP    NOT NULL,
    published_at   TIMESTAMP
);
CREATE INDEX idx_outbox_unpublished ON outbox_event (created_at) WHERE published_at IS NULL;

Partial index 只包含未发送的行。这样 relay 的查询始终很快，即使表中已有数百万行旧数据。

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    private UUID id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    @Column(columnDefinition = "text")
    private String payload;
    private Instant createdAt;
    private Instant publishedAt;

    protected OutboxEvent() {
    }

    public OutboxEvent(String aggregateType, String aggregateId, String eventType, String payload) {
        this.id = UUID.randomUUID();
        this.aggregateType = aggregateType;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = Instant.now();
    }

    public void markPublished(Instant at) {
        this.publishedAt = at;
    }

    public UUID getId() { return id; }
    public String getAggregateId() { return aggregateId; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public Instant getPublishedAt() { return publishedAt; }
}

id 也充当 event id 的唯一标识。Consumer 将使用它来过滤重复（第 6 节）。aggregateId 将是 Kafka 消息的 key。

4.3. 在同一 transaction 中写入 Order 和事件

public record OrderPlacedEvent(Long orderId, String customerId, BigDecimal amount) {
}

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public OrderService(OrderRepository orderRepository,
                        OutboxEventRepository outboxRepository,
                        ObjectMapper objectMapper) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public Order placeOrder(String customerId, BigDecimal amount) {
        Order order = orderRepository.save(new Order(customerId, amount));

        var event = new OrderPlacedEvent(order.getId(), customerId, amount);
        outboxRepository.save(new OutboxEvent(
                "Order", order.getId().toString(), "OrderPlaced", toJson(event)));

        return order;
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("不能序列化事件", e);
        }
    }
}

关键点在于 @Transactional。两个 save 命令要么一起 commit，要么一起 rollback。OrderService 不再知道 Kafka，所以 transaction 中没有网络调用。

4.4. Message relay 推送事件到 Kafka

首先，写一个 repository 有行锁查询：

public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    @Query(value = """
            SELECT * FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT :batchSize
            FOR UPDATE SKIP LOCKED
            """, nativeQuery = true)
    List<OutboxEvent> findBatchToPublish(@Param("batchSize") int batchSize);
}

FOR UPDATE SKIP LOCKED（PostgreSQL、MySQL 8+）允许多个应用实例同时运行。每个实例锁定自己的一批，并跳过其他实例正在处理的行。这样两个实例就不会并行发送同一个事件。

接下来是 relay：

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);

    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(OutboxEventRepository outboxRepository,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPending() {
        for (OutboxEvent event : outboxRepository.findBatchToPublish(100)) {
            try {
                kafkaTemplate.send(toRecord(event)).get(5, TimeUnit.SECONDS);
                event.markPublished(Instant.now());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                log.warn("发送事件 {} 失败，将重试", event.getId(), e);
                break; // 停止以保持顺序，下次运行时重试
            }
        }
    }

    private ProducerRecord<String, String> toRecord(OutboxEvent event) {
        var record = new ProducerRecord<>("orders", event.getAggregateId(), event.getPayload());
        record.headers()
              .add("eventId", event.getId().toString().getBytes(StandardCharsets.UTF_8))
              .add("eventType", event.getEventType().getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

逐段解释：

@Scheduled(fixedDelay = 1000) 在上一次运行结束一秒后运行 relay。记得在配置类上添加 @EnableScheduling。
.get(5, TimeUnit.SECONDS) 等待 broker 确认。Relay 只有在确定 Kafka 已收到消息时才标记 publishedAt。
markPublished 修改被管理的 entity。JPA 在 transaction commit 时自动生成 UPDATE 命令，你不需要调用 save。
出错时，relay break 而不是跳过该行。因此同一订单的事件不会被打乱顺序。
Key 是 aggregateId，所以一个 Order 的所有事件都落在同一分区并在 Kafka 上保持顺序。

当 broker 暂时停止时，日志看起来像这样：

WARN  OutboxRelay : 发送事件 3f6c1e0a-... 失败，将重试
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

当 broker 恢复时，下一次运行将继续发送该事件。

注意： 如果应用在 Kafka 已收到消息但 transaction 提交之前崩溃，outbox 行仍然是“未发送”。下一次运行将重新发送。这正是 at-least-once，consumer 必须能处理重复。

5. 测试

第一个测试检查每个订单是否生成了一行 outbox：

@DataJpaTest
@Import({OrderService.class, JacksonAutoConfiguration.class})
class OrderServiceTest {

    @Autowired
    OrderService orderService;

    @Autowired
    OutboxEventRepository outboxRepository;

    @Test
    void placeOrder_shouldWriteOutboxEventInSameTransaction() {
        Order order = orderService.placeOrder("customer-42", new BigDecimal("150000"));

        List<OutboxEvent> events = outboxRepository.findAll();

        assertThat(events).hasSize(1);
        OutboxEvent event = events.getFirst();
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.getEventType()).isEqualTo("OrderPlaced");
        assertThat(event.getPayload()).contains("\"customerId\":\"customer-42\"");
        assertThat(event.getPublishedAt()).isNull();
    }
}

Repository 的本地查询在启动时不被检查。因此 H2 仍然可以运行这个测试，尽管它不完全支持 SKIP LOCKED。要用真实查询测试 relay，你应该通过 Testcontainers 使用 PostgreSQL。

第二个测试确保 Kafka 出错时 relay 不标记为已发送：

class OutboxRelayTest {

    OutboxEventRepository repository = mock(OutboxEventRepository.class);
    KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
    OutboxRelay relay = new OutboxRelay(repository, kafkaTemplate);

    @Test
    void shouldKeepEventUnpublished_whenKafkaFails() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker down")));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNull();
    }

    @Test
    void shouldMarkPublished_whenKafkaAcknowledges() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.completedFuture(null));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNotNull();
    }
}

这两个测试锁定了核心行为：只有当 broker 已确认时才标记为已发送。

6. Consumer 端：幂等处理

由于可能收到重复，consumer 需要记住已处理的 eventId。这个记忆必须与业务逻辑在同一 transaction 中：

@Component
public class InventoryListener {

    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public InventoryListener(ProcessedEventRepository processedEvents,
                             InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    @Transactional
    public void onOrderPlaced(@Header("eventId") String eventId, String payload) {
        if (processedEvents.existsById(eventId)) {
            return; // 已处理，跳过重复
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

processed_event 表应该以 event_id 作为主键。如果两个线程同时处理一个重复，约束将阻止其中之一。

7. 常见错误和何时不应该使用

在 OrderService 中调用 Kafka“加快速度”。 这样做就回到了 dual write。所有事件都必须经过 outbox 表。
忘记清理 outbox 表。 表会很快膨胀。应该有一个 job 删除 published_at 早于几天的行。
跳过错误的行然后继续发送。 这会破坏事件顺序。如果 payload 永久损坏，将其转移到 FAILED 状态供调查，不要无限重试。
轮询过于频繁。 100 ms 的周期与多个实例对数据库产生相当的负载。0.5-2 秒的周期通常足够。
需要非常低的延迟或非常高的吞吐量。 此时应将轮询替换为 CDC（变更数据捕获）。例如 Debezium 读取 WAL/binlog，然后直接将 outbox 行推送到 Kafka。Outbox 表保持不变，只有 relay 改变。

注意： 如果事件不重要，不要使用 Outbox 模式，例如指标或用户行为日志可以丢失。此时直接发送到 Kafka 要简单得多。反过来说，Kafka transactions (transactional.id) 不能替代 Outbox：它们只保证 Kafka topic 之间的原子性，不包括你的数据库。

8. 结论

数据库和 Kafka 之间的 dual write 总是有漏洞导致事件丢失或重复。Outbox 模式通过三个步骤堵住这个漏洞。首先，在与业务数据相同的 transaction 中将事件写入 outbox 表。其次，使用单独的 relay 将事件推送到 Kafka，只有在 broker 确认时才标记为已发送。第三，在 consumer 处进行幂等处理以吸收 at-least-once 语义的重复。

你现在拥有所有的组件：带 partial index 的 schema、不依赖 Kafka 的 OrderService、使用 FOR UPDATE SKIP LOCKED 的 relay 以支持多个实例、核心行为测试和防重复的 consumer。当系统增大时，你可以将 relay 轮询替换为 CDC，而无需修改业务代码。

使用 Kafka 的 Outbox 模式：数据库写入时不丢失事件

1. 概览

2. Dual write 问题

3. Outbox 模式的思想

4. 用 Spring Boot 实现

4.1. 依赖

4.2. Outbox 表和 entity

4.3. 在同一 transaction 中写入 Order 和事件

4.4. Message relay 推送事件到 Kafka

5. 测试

6. Consumer 端：幂等处理

7. 常见错误和何时不应该使用

8. 结论

同主题文章

Kafka 中的 Consumer Group 与 Rebalance 在 Spring Boot 中的应用

Kafka producer：acks、幂等性和批处理

Saga pattern 与 Kafka 分布式事务：预订旅游套餐示例

Schema Registry 和 Avro/Protobuf 在 Kafka 与 Spring Boot 中的应用

事件溯源和 Kafka：保存事件而非状态