如何用 Spring Cloud Stream、Kafka 实现事件驱动微服务
结论
MariaDB 本地事务和 Kafka 事务是两个独立的事务资源。仅使用 @Transactional、Spring Cloud Stream 和 Kafka，通常无法让“写订单数据库”和“发布 Kafka 消息”组成严格的原子事务。
生产环境中，较稳妥的做法是采用 Transactional Outbox（事务发件箱）模式：
- 在同一个 MariaDB 本地事务中写入订单和 Outbox 事件。
- 事务提交后，由独立发布器或 CDC 工具将 Outbox 事件发送到 Kafka。
- 发布成功后，将事件标记为已发送。
- Payment Service 按至少一次投递处理消息，并通过事件 ID 实现幂等消费。
orderSource.output().send(...) 返回 true，只能说明 Spring 消息通道接受了消息，不能说明 Kafka broker 已确认写入。要确认 Kafka 的发送结果，需要使用生产者的异步回调或等待发送结果，并设置合适的 acks。但即使收到了 Kafka 的确认，也无法解决数据库与 Kafka 之间的原子性问题。
为什么不能直接放进一个 @Transactional
下面的代码涉及两个不同的事务资源：
@Transactional
public Order createOrder(Order order) {
orderRepository.save(order);
orderSource.output()
.send(MessageBuilder.withPayload(order).build());
return order;
}MariaDB 事务由数据库事务管理器控制，Kafka 消息则由 Kafka producer 发送。执行过程中可能遇到两个故障窗口：
- 数据库已经提交，但 Kafka 消息发送失败。订单存在，Payment Service 却始终收不到消息。
- Kafka 消息已经提交，但数据库随后回滚。Payment Service 收到了一个实际不存在的订单。
改变两步操作的执行顺序不能消除问题，只会改变可能出现的不一致类型。
理论上可以使用 XA/JTA 两阶段提交，但 Kafka 与关系数据库的组合通常不会选择这种方案。它的配置复杂，性能和可用性成本较高，具体支持程度还取决于驱动、框架和版本。微服务架构一般采用最终一致性，而不是为数据库和消息系统建立分布式 ACID 事务。
推荐方案：Transactional Outbox
先在 Order Service 的数据库中增加一张事件表：
CREATE TABLE outbox_event (
id VARCHAR(36) NOT NULL PRIMARY KEY,
aggregate_type VARCHAR(100) NOT NULL,
aggregate_id VARCHAR(100) NOT NULL,
event_type VARCHAR(100) NOT NULL,
payload LONGTEXT NOT NULL,
status VARCHAR(20) NOT NULL,
created_at TIMESTAMP(6) NOT NULL,
published_at TIMESTAMP(6) NULL
);
CREATE INDEX idx_outbox_status_created
ON outbox_event(status, created_at);创建订单时，不再直接向 Kafka 发送消息，而是在同一个数据库事务中保存订单和 Outbox 记录：
@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 createOrder(Order order) throws JsonProcessingException {
Order savedOrder = orderRepository.save(order);
OrderCreatedEvent event = new OrderCreatedEvent(
UUID.randomUUID().toString(),
savedOrder.getId(),
savedOrder.getAmount()
);
OutboxEvent outboxEvent = new OutboxEvent();
outboxEvent.setId(event.eventId());
outboxEvent.setAggregateType("Order");
outboxEvent.setAggregateId(savedOrder.getId().toString());
outboxEvent.setEventType("OrderCreated");
outboxEvent.setPayload(objectMapper.writeValueAsString(event));
outboxEvent.setStatus("NEW");
outboxEvent.setCreatedAt(Instant.now());
outboxRepository.save(outboxEvent);
return savedOrder;
}
}事件对象需要包含稳定且唯一的事件 ID：
public record OrderCreatedEvent(
String eventId,
Long orderId,
BigDecimal amount
) {
}Controller 只负责调用业务服务：
@PostMapping
public Order createOrder(@RequestBody Order order) throws JsonProcessingException {
return orderService.createOrder(order);
}这样，事务成功时，订单和待发布事件会同时存在；事务回滚时，两者都不会写入数据库。数据库中不会再出现“订单已经保存，但没有对应待发送事件”的状态。
将 Outbox 事件发布到 Kafka
发布 Outbox 事件通常有两种方式。
方式一：CDC
可以使用 Debezium 等 CDC 工具监听数据库变更，再把 Outbox 表中的新增记录发送到 Kafka。
这种方式不需要应用定时扫描数据库，延迟通常也较低，但需要额外部署和维护 CDC、Kafka Connect 等组件。具体配置取决于数据库、连接器和版本。
方式二：应用轮询发布
也可以由 Order Service 定期查询未发布的事件并发送：
@Component
public class OutboxPublisher {
private final OutboxEventRepository outboxRepository;
private final KafkaTemplate<String, String> kafkaTemplate;
public OutboxPublisher(
OutboxEventRepository outboxRepository,
KafkaTemplate<String, String> kafkaTemplate) {
this.outboxRepository = outboxRepository;
this.kafkaTemplate = kafkaTemplate;
}
@Scheduled(fixedDelayString = "${outbox.publish-delay-ms:1000}")
public void publish() {
List<OutboxEvent> events =
outboxRepository.findTop100ByStatusOrderByCreatedAt("NEW");
for (OutboxEvent event : events) {
try {
kafkaTemplate
.send("order-created", event.getAggregateId(), event.getPayload())
.get();
markAsPublished(event.getId());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
} catch (ExecutionException e) {
// 保留 NEW 状态，稍后重试
log.warn("Failed to publish outbox event {}", event.getId(), e);
}
}
}
@Transactional
public void markAsPublished(String eventId) {
outboxRepository.markPublished(
eventId,
"PUBLISHED",
Instant.now()
);
}
}这里的 .get() 会等待 Kafka producer 返回成功或失败结果。实际项目也可以使用异步回调来提高吞吐量。
轮询发布仍有一个无法完全消除的故障窗口：
- Kafka 已成功接收消息。
- 应用在更新 Outbox 状态之前崩溃。
- 应用重启后，再次发送同一个事件。
因此，Outbox 通常提供的是 至少一次投递。消费者必须能够安全地处理重复消息。
如果多个实例同时扫描 Outbox，还要通过行锁、状态抢占或 SELECT ... FOR UPDATE SKIP LOCKED 等机制，避免不同实例反复处理同一批记录。可使用的具体 SQL 能力取决于 MariaDB 版本。
Payment Service 必须实现幂等消费
Payment Service 可以建立一张已处理事件表，并把“记录事件”和“创建支付记录”放在同一个本地事务中：
CREATE TABLE processed_event (
event_id VARCHAR(36) NOT NULL PRIMARY KEY,
processed_at TIMESTAMP(6) NOT NULL
);消费代码示例如下：
@Service
public class PaymentEventHandler {
private final ProcessedEventRepository processedEventRepository;
private final PaymentRepository paymentRepository;
public PaymentEventHandler(
ProcessedEventRepository processedEventRepository,
PaymentRepository paymentRepository) {
this.processedEventRepository = processedEventRepository;
this.paymentRepository = paymentRepository;
}
@Transactional
public void handle(OrderCreatedEvent event) {
if (processedEventRepository.existsById(event.eventId())) {
return;
}
Payment payment = new Payment();
payment.setOrderId(event.orderId());
payment.setAmount(event.amount());
payment.setStatus("PENDING");
paymentRepository.save(payment);
processedEventRepository.save(
new ProcessedEvent(event.eventId(), Instant.now())
);
}
}并发控制必须依靠 event_id 的唯一约束，不能只依赖先查询、再插入。两个消费者可能同时查到记录“不存在”，最后仍要由数据库的唯一约束阻止重复处理。
如果“收费”操作还会调用外部支付机构，则需要使用支付机构提供的幂等键，例如将 orderId 或独立的支付请求 ID 作为幂等标识。只保证本地数据库幂等，并不能防止外部机构重复扣款。
如何确认消息已经到达 Kafka
下面这段代码：
orderSource.output()
.send(MessageBuilder.withPayload(order).build());它的返回值遵循 Spring Messaging 的 MessageChannel.send(...) 语义。返回 true 通常只能说明通道接受了消息，不能将其视为 Kafka broker 的持久化确认。
具体的确认方式取决于 Spring Cloud Stream 和 Spring Kafka 的版本。较直接的做法是处理 KafkaTemplate 返回的异步结果：
CompletableFuture<SendResult<String, OrderCreatedEvent>> future =
kafkaTemplate.send("order-created", orderId, event);
future.whenComplete((result, exception) -> {
if (exception != null) {
log.error("Kafka publish failed, eventId={}", event.eventId(), exception);
return;
}
RecordMetadata metadata = result.getRecordMetadata();
log.info(
"Kafka publish succeeded, topic={}, partition={}, offset={}",
metadata.topic(),
metadata.partition(),
metadata.offset()
);
});如果当前 Spring Kafka 版本返回 ListenableFuture，就需要使用对应版本的回调 API，不能混用不同版本的代码示例。
确实需要同步等待时，可以这样处理：
try {
SendResult<String, OrderCreatedEvent> result =
kafkaTemplate.send("order-created", orderId, event).get();
RecordMetadata metadata = result.getRecordMetadata();
log.info(
"Published to partition {} at offset {}",
metadata.partition(),
metadata.offset()
);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Kafka send interrupted", e);
} catch (ExecutionException e) {
throw new IllegalStateException("Kafka send failed", e.getCause());
}同步等待会增加请求延迟并占用工作线程，因此通常更适合放在 Outbox 后台发布器中，而不是直接放进 HTTP 请求链路。
Producer 的确认配置
Producer 需要根据业务的可靠性要求设置确认级别。常见配置如下：
spring:
kafka:
producer:
acks: all
properties:
enable.idempotence: true如果使用 Spring Cloud Stream Kafka binder，应将这些属性放在当前版本规定的 binder 或 producer 配置位置。不同版本的 Spring Cloud Stream 可能使用不同的属性结构，需要以项目所用版本的配置参考为准。
这些配置和结果分别表示：
acks=all：按照 Kafka 当前的 ISR 规则，等待所有同步副本确认。enable.idempotence=true：减少 producer 重试造成的重复写入。- 发送 future 成功：Kafka producer 已根据
acks配置收到 broker 确认。 - 发送 future 失败：发送操作未能在重试和超时策略规定的范围内完成。
即使 future 返回失败，也不能断定 broker 一定没有写入消息。例如，broker 已保存记录，但确认响应在网络中丢失，此时客户端仍可能看到超时。因此，业务层仍要使用事件 ID，并保证消费操作幂等。
Kafka 事务能解决什么
Kafka 事务可以原子提交多条 Kafka 记录，也可以与消费者配合，实现“消费 Kafka 消息并产生新的 Kafka 消息”的 read-process-write 原子流程。
但它不能直接将 MariaDB 提交和 Kafka 提交合并成一个完全原子的事务。Spring 的事务同步或组合事务管理方式可以协调提交顺序，可进程仍可能在第一个资源提交后、第二个资源提交前崩溃。这类方案只能缩小故障窗口，无法保证不发生不一致。
这个场景更适合使用下面的事务边界：
HTTP 请求
└── MariaDB 本地事务
├── INSERT orders
└── INSERT outbox_event
Outbox 发布器或 CDC
└── Kafka order-created
Payment Service
└── MariaDB 本地事务
├── INSERT payment
└── INSERT processed_event
付款失败时，通常会继续发布 PaymentFailed 事件，由 Order Service 将订单更新为失败或取消状态。付款成功时则发布 PaymentCompleted。这是 Saga 的事件编排方式，目标是实现可恢复、可追踪的最终一致性，而不是把多个微服务数据库包装成一个跨系统 ACID 事务。
备注：内容仅供参考。