spring-boot-event-driven-patterns

Spring Boot 事件驱动模式

概述

在 Spring Boot 3.x 中，使用领域事件、ApplicationEventPublisher、@TransactionalEventListener，以及通过 Kafka 和 Spring Cloud Stream 实现的分布式消息传递，构建事件驱动架构（EDA）模式。

适用场景

• 使用 Kafka 消息传递实现事件驱动微服务
• 在 DDD 架构中从聚合根发布领域事件
• 设置在数据库事务提交后触发的事务事件监听器
• 通过 Spring Kafka 添加带有生产者和消费者的异步消息传递
• 使用事务发件箱模式确保事件可靠送达
• 用基于事件的服务间通信替代同步调用

快速参考

概念	描述
领域事件	不可变事件，扩展 DomainEvent 基类，包含 eventId、occurredAt、correlationId
事件发布	本地发布使用 ApplicationEventPublisher.publishEvent()，分布式发布使用 KafkaTemplate
事件监听	使用 @TransactionalEventListener(phase = AFTER_COMMIT) 可靠地处理事件
Kafka	使用 @KafkaListener(topics = "...") 消费分布式事件
Spring Cloud Stream	采用 Consumer Bean 的函数式编程模型
发件箱模式	将事件与业务数据原子化存储，并使用定时发布器

示例

将单体应用重构为事件驱动架构

之前（反模式）：

@Transactional
public Order processOrder(OrderRequest request) {
    Order order = orderRepository.save(request);
    inventoryService.reserve(order.getItems()); // Blocking
    paymentService.charge(order.getPayment()); // Blocking
    emailService.sendConfirmation(order); // Blocking
    return order;
}

之后（事件驱动）：

@Transactional
public Order processOrder(OrderRequest request) {
    Order order = Order.create(request);
    orderRepository.save(order);

    // Publish event after transaction commits
    eventPublisher.publishEvent(new OrderCreatedEvent(order.getId(), order.getItems()));

    return order;
}

@Component
public class OrderEventHandler {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void handleOrderCreated(OrderCreatedEvent event) {
        // Execute asynchronously after the order is saved
        inventoryService.reserve(event.getItems());
        paymentService.charge(event.getPayment());
    }
}

完整的可运行示例请参见 examples.md。

说明

1. 设计领域事件

创建扩展 DomainEvent 基类的不可变事件类：

public abstract class DomainEvent {
    private final UUID eventId;
    private final LocalDateTime occurredAt;
    private final UUID correlationId;
}

public class ProductCreatedEvent extends DomainEvent {
    private final ProductId productId;
    private final String name;
    private final BigDecimal price;
}

模式请参见 domain-events-design.md。

2. 从聚合中发布事件

将领域事件添加到聚合根，并通过 ApplicationEventPublisher 发布：

@Service
@Transactional
public class ProductService {
    public Product createProduct(CreateProductRequest request) {
        Product product = Product.create(request.getName(), request.getPrice(), request.getStock());
        repository.save(product);

        product.getDomainEvents().forEach(eventPublisher::publishEvent);
        product.clearDomainEvents();

        return product;
    }
}

DDD 模式请参见 aggregate-root-patterns.md。

3. 以事务方式处理事件

使用 @TransactionalEventListener 可靠地处理事件：

@Component
public class ProductEventHandler {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onProductCreated(ProductCreatedEvent event) {
        notificationService.sendProductCreatedNotification(event.getName());
    }
}

验证： 检查处理器执行前数据库状态是否已提交，以确认事件处理器仅在事务提交后触发。

事件处理模式请参见 event-handling.md。

4. 配置 Kafka 基础设施

配置 KafkaTemplate 以发布消息，使用 @KafkaListener 消费消息：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

验证： 通过 KafkaTemplate 发送测试事件，并确认消费者日志中出现该事件，然后再继续采用生产环境模式。

请参见 dependency-setup.md 和 configuration.md。

5. 实现发件箱模式

创建 OutboxEvent 实体以原子化存储事件：

@Entity
public class OutboxEvent {
    private UUID id;
    private String aggregateId;
    private String eventType;
    private String payload;
    private LocalDateTime publishedAt;
}

验证： 检查定时任务运行后是否设置了 publishedAt 时间戳，以确认定时处理器已获取待处理事件。

定时处理器会发布待处理事件。请参见 outbox-pattern.md。

6. 处理故障场景

实现重试逻辑、死信队列和幂等处理器：

@RetryableTopic(attempts = "3")
@KafkaListener(topics = "product-events")
public void handleProductEvent(ProductCreatedEventDto event) {
    orderService.onProductCreated(event);
}

验证： 在进行可观测性配置前，确认消息在重试次数耗尽后到达死信主题。

7. 添加可观测性

启用 Spring Cloud Sleuth 进行分布式追踪，并监控指标。

最佳实践

• 使用过去时命名：ProductCreated（而非 CreateProduct）
• 保持事件不可变：所有字段都应为 final
• 包含关联 ID：用于跨服务追踪事件
• 使用 AFTER_COMMIT 阶段：确保事件在数据库事务成功提交后发布
• 实现幂等处理程序：妥善处理重复事件
• 添加重试机制：处理失败的事件，并使用指数退避
• 实现死信队列：处理重试后仍失败的事件
• 记录所有故障：提供充足的上下文以便调试
• 确保处理程序不依赖顺序：分布式系统无法保证事件顺序
• 批量处理事件：处理大量事件时采用
• 监控事件延迟：为处理缓慢设置告警

参考资料

• dependency-setup.md — Maven/Gradle 依赖
• configuration.md — Kafka 和 Spring Cloud Stream 配置
• domain-events-design.md — 领域事件设计模式
• aggregate-root-patterns.md — 带事件发布功能的聚合根
• event-publishing.md — 本地和分布式事件发布
• event-handling.md — 事件处理与消费模式
• outbox-pattern.md — 用于提高可靠性的事务发件箱模式
• testing-strategies.md — 单元测试和集成测试方法
• examples.md — 完整的可运行示例
• event-driven-patterns-reference.md — 详细参考文档

限制和警告

• 使用 @TransactionalEventListener 发布的事件只会在事务提交后触发
• 避免在事件中发布大型对象（可能造成内存压力和序列化问题）
• 谨慎使用异步事件处理程序（可能涉及独立线程和并发问题）
• Kafka 消费者必须处理重复消息（实现幂等处理）
• 分布式系统无法保证事件顺序（设计不依赖顺序的处理程序）
• 切勿在主事务线程的事件监听器中执行阻塞操作
• 监控事件处理积压情况（这可能表明系统容量存在问题）

相关技能

• spring-boot-security-jwt — 用于安全事件发布的 JWT 身份验证
• spring-boot-test-patterns — 测试事件驱动应用程序
• aws-sdk-java-v2-lambda — 使用 AWS Lambda 进行事件驱动处理
• langchain4j-tool-function-calling-patterns — AI 驱动的事件处理