# Spring Boot 事件驱动模式 ## 概述使用域事件、ApplicationEventPublisher、@TransactionalEventListener 以及使用 Kafka 和 Spring Cloud Stream 进行分布式消息传递，在 Spring Boot 3.x 中实现事件驱动架构 (EDA) 模式。 ## 何时使用此技能在构建需要以下功能的应用程序时使用此技能： * 通过基于事件的通信实现微服务之间的松耦合 * DDD架构中聚合根的领域事件发布 * 事务事件监听器确保数据库提交后的一致性 * 使用 Kafka 进行分布式消息传递以实现服务间通信 * 使用 Spring Cloud Stream 进行反应式系统的事件流处理 * 使用事务发件箱模式的可靠性 * 有界上下文之间的异步通信 * 具有适当事件溯源模式的事件溯源基础 ## 说明按照以下步骤在 Spring Boot 中实现事件驱动架构模式： ### 1\.设计领域活动创建扩展 DomainEvent 基类的不可变事件类。包含 eventId、occurredAt 和correlationId 字段以实现可追溯性。 ### 2\.定义事件发布将ApplicationEventPublisher添加到需要发布事件的服务中。域状态更改完成后发布事件。 ### 3\.配置事务监听器使用@TransactionalEventListener 和phase = AFTER_COMMIT 确保仅在成功的数据库事务后处理事件。 ### 4\.设置 Kafka 基础设施配置 KafkaTemplate 以将事件发布到主题。创建 @KafkaListener bean 来使用来自其他服务的事件。 ### 5\.实施 Spring Cloud Stream 使用带有 Consumer bean 定义的函数式编程模型来进行反应式事件消费。在 application.yml 中配置绑定。 ### 6\.处理失败场景使用指数退避实现重试逻辑。为失败的消息配置死信队列。使事件处理程序具有幂等性。 ### 7\.实施发件箱模式创建 OutboxEvent 实体以自动存储事件和业务数据。使用计划作业将发件箱事件发布到消息代理。 ### 8\.添加可观察性启用 Spring Cloud Sleuth 进行分布式跟踪。通过 Actuator 端点监视事件处理指标。 ## 设置和配置 ### 所需的依赖项要实现事件驱动模式，请在项目中包含以下依赖项： **马夫：** <依赖关系> <依赖关系> org.springframework.boot spring-boot-starter-web <依赖关系> org.springframework.boot spring-boot-starter-data-jpa <依赖关系> org.springframework.kafka spring-kafka <依赖关系> org.springframework.cloud spring-cloud-stream 4.0.4 // 使用最新兼容版本 <依赖关系> org.springframework.boot spring-boot-starter-test <范围>测试 <依赖关系> org.testcontainers 测试容器 <版本>1.19.0 <范围>测试 **摇篮：** 依赖项{ // Spring Boot 网络实现 'org.springframework.boot:spring-boot-starter-web' // Spring数据JPA 实现 'org.springframework.boot:spring-boot-starter-data-jpa' // 卡夫卡实现 'org.springframework.kafka:spring-kafka' // Spring 云流实现 'org.springframework.cloud:spring-cloud-stream:4.0.4' // 测试 testImplementation 'org.springframework.boot:spring-boot-starter-test' testImplementation 'org.testcontainers:testcontainers:1.19.0' } ### 基本配置配置您的应用程序以实现事件驱动架构： # 服务器配置服务器端口=8080 # 卡夫卡配置 spring.kafka.bootstrap-servers=localhost:9092 spring.kafka. Producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer spring.kafka. Producer.value-serializer = org.springframework.kafka.support.serializer.JsonSerializer # Spring Cloud Stream 配置 spring.cloud.stream.kafka.binder.brokers=localhost:9092 ## 核心模式 ### 1\.领域活动设计为业务域更改创建不可变域事件： // 领域事件基类公共抽象类DomainEvent { 私有最终 UUID eventId；私人最终 LocalDateTime 发生于；私有最终UUID相关Id；受保护的域事件() { this.eventId = UUID.randomUUID(); this.occurredAt = LocalDateTime.now(); this.correlationId = UUID.randomUUID(); } 受保护的DomainEvent（UUID相关Id）{ this.eventId = UUID.randomUUID(); this.occurredAt = LocalDateTime.now(); this.correlationId = correlationId; } // 吸气剂公共 UUID getEventId() { return eventId; } 公共LocalDateTime getOccurredAt() { 返回发生时间; } 公共UUID getCorrelationId() { 返回correlationId; } } // 特定领域事件公共类 ProductCreatedEvent 扩展 DomainEvent { 私有最终 ProductId 产品Id；私有最终字符串名称；私人最终 BigDecimal 价格；私人最终整数股票； public ProductCreatedEvent(ProductId ProductId, 字符串名称, BigDecimal 价格, 整数股票) { 超级（）； this.productId = 产品Id; this.name = 名称；这个.价格 = 价格; this.stock = 股票； } // 吸气剂公共 ProductId getProductId() { return ProductId; } 公共字符串 getName() { 返回名称; } 公共 BigDecimal getPrice() { 返回价格; } public Integer getStock() { 返回股票； } } ### 2\.具有事件发布的聚合根实现发布域事件的聚合： @实体 @盖特 @ToString @EqualsAndHashCode(of = "id") @NoArgsConstructor（访问= AccessLevel.PROTECTED）公开课产品{ @Id 私有 ProductId id；私有字符串名称；私人 BigDecimal 价格；私人整数股票； @瞬态 private List domainEvents = new ArrayList<>(); 公共静态产品创建（字符串名称，BigDecimal价格，整数股票）{ 产品产品 = 新产品(); 产品.id = ProductId.generate(); 产品名称=名称；产品.价格=价格；产品.库存=库存； Product.domainEvents.add(new ProductCreatedEvent(product.id, 名称, 价格, 库存)); 退回产品； } 公共无效减少股票（整数数量）{ this.stock -= 数量； this.domainEvents.add(new ProductStockDecreasedEvent(this.id, 数量, this.stock)); } 公共列表 getDomainEvents() { 返回新的ArrayList<>(domainEvents); } 公共无效clearDomainEvents（）{ 域事件.clear(); } } ### 3\.应用事件发布从应用程序服务发布域事件： @服务 @RequiredArgsConstructor @事务性公共类 ProductApplicationService { 私有最终 ProductRepository 产品存储库；私有最终ApplicationEventPublisher eventPublisher；公共ProductResponse createProduct（CreateProductRequest请求）{ 产品产品 = Product.create( 请求.getName(), 请求.getPrice(), request.getStock() ）；产品存储库.保存（产品）； // 发布领域事件产品.getDomainEvents().forEach(eventPublisher::publishEvent); 产品.clearDomainEvents(); 返回mapToResponse（产品）； } } ### 4\.本地事件处理使用事务事件监听器处理事件： @组件 @RequiredArgsConstructor 公共类 ProductEventHandler { 私有最终NotificationService notificationService；私人最终审计服务审计服务； @TransactionalEventListener(阶段 = TransactionPhase.AFTER_COMMIT) 公共无效onProductCreated（ProductCreatedEvent事件）{ 审计服务.logProductCreation( event.getProductId().getValue(), 事件.getName(), 事件.getPrice(), event.getCorrelationId() ）； notificationService.sendProductCreatedNotification(event.getName()); } @TransactionalEventListener(阶段 = TransactionPhase.AFTER_COMMIT) 公共无效onProductStockDecreased（ProductStockDecreasedEvent事件）{ notificationService.sendStockUpdateNotification( event.getProductId().getValue(), event.getQuantity() ）； } } ### 5\.分布式事件发布向 Kafka 发布事件以进行服务间通信： @组件 @RequiredArgsConstructor 公共类 ProductEventPublisher { 私有最终 KafkaTemplate kafkaTemplate; 公共无效publishProductCreatedEvent（ProductCreatedEvent事件）{ ProductCreatedEventDto dto = mapToDto(事件); kafkaTemplate.send("产品事件", event.getProductId().getValue(), dto); } 私人ProductCreatedEventDto mapToDto（ProductCreatedEvent事件）{ 返回新的 ProductCreatedEventDto( event.getEventId(), event.getProductId().getValue(), 事件.getName(), 事件.getPrice(), 事件.getStock(), event.getOccurredAt(), event.getCorrelationId() ）； } } ### 6\.使用 Spring Cloud Stream 的事件消费者使用函数式编程风格使用事件： @组件 @RequiredArgsConstructor 公共类 ProductEventStreamConsumer { 私有最终 OrderService orderService; @豆子公共 Consumer ProductCreatedConsumer() { 返回事件 -> { orderService.onProductCreated(事件); }; } @豆子公共 Consumer ProductStockDecreasedConsumer() { 返回事件 -> { orderService.onProductStockDecreased(事件); }; } } ## 高级模式 ### 事务发件箱模式使用发件箱模式确保可靠的事件发布： @实体 @Table(名称 = "outbox_events") @盖特 @塞特 @Builder @NoArgs构造函数 @AllArgsConstructor 公共类 OutboxEvent { @Id @GenerateValue(策略 = GenerationType.UUID) 私有 UUID ID；私有字符串聚合Id；私有字符串事件类型；私有字符串负载；私有UUID相关Id；私有 LocalDateTime 创建于；私人 LocalDateTime 发布于；私有整数重试计数； } @组件 @RequiredArgsConstructor 公共类 OutboxEventProcessor { 私有最终 OutboxEventRepository outboxRepository; 私有最终 KafkaTemplate kafkaTemplate; @Scheduled(固定延迟 = 5000) @事务性公共无效processPendingEvents（）{ ListendingEvents = outboxRepository.findByPublishedAtNull(); for (OutboxEvent 事件：pendingEvents) { 尝试{ kafkaTemplate.send("产品事件", event.getAggregateId(), event.getPayload()); event.setPublishedAt(LocalDateTime.now()); outboxRepository.save(事件); } catch (异常 e) { event.setRetryCount(event.getRetryCount() + 1); outboxRepository.save(事件); } } } } ## 测试策略 ### 单元测试领域事件测试域事件发布和处理：类产品测试{ @测试无效shouldPublishProductCreatedEventOnCreation（）{ 产品产品 = Product.create("测试产品", BigDecimal.TEN, 100); assertThat(product.getDomainEvents()).hasSize(1); assertThat(product.getDomainEvents().get(0)) .isInstanceOf(ProductCreatedEvent.class); } } @ExtendWith(MockitoExtension.class) 类 ProductEventHandlerTest { @模拟私有NotificationService通知服务； @InjectMocks 私有 ProductEventHandler 处理程序； @测试无效shouldHandleProductCreatedEvent（）{ ProductCreatedEvent 事件 = new ProductCreatedEvent( ProductId.of("123"), "产品", BigDecimal.TEN, 100 ）； handler.onProductCreated(事件); 验证（notificationService）。sendProductCreatedNotification（“产品”）； } } ### 使用测试容器进行集成测试测试 Kafka 与 Testcontainers 的集成： @SpringBootTest @测试容器类 KafkaEventIntegrationTest { @容器静态 KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluenceinc/cp-kafka:7.5.0")); @Autowired 私有 ProductApplicationService 产品服务； @DynamicPropertySource 静态无效configureProperties（DynamicPropertyRegistry注册表）{ registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers); } @测试无效shouldPublishEventToKafka（）{ CreateProductRequest 请求 = new CreateProductRequest( “测试产品”，BigDecimal.valueOf(99.99)，50 ）； ProductResponse 响应 = ProductService.createProduct(request); // 验证事件已发布验证(eventPublisher).publishProductCreatedEvent(any(ProductCreatedEvent.class)); } } ## 最佳实践 ### 活动设计指南 * **使用过去时命名** ：ProductCreated，而不是CreateProduct * **保持事件不可变**：所有字段都应该是最终的 * **包括相关 ID** ：用于跨服务跟踪事件 * **序列化为 JSON** ：为了跨服务兼容性 ### 事务一致性 * **使用 AFTER_COMMIT 阶段** ：确保在成功的数据库事务后发布事件 * **实现幂等处理程序**：优雅地处理重复事件 * **添加重试机制** : 对于失败的事件处理 ### 错误处理 * **实现死信队列** ：对于处理失败的事件 * **记录所有失败**：包括足够的调试上下文 * **设置适当的超时**：用于事件处理操作 ### 性能考虑因素 * **批量事件处理**：处理大量事件时 * **使用正确的分区**：对于 Kafka 主题 * **监控事件延迟**：设置处理缓慢的警报 ## 示例和参考有关综合示例，请参阅以下资源： * [完整的工作示例]() * [详细实现模式]() ## 故障排除 ### 常见问题 **未发布的事件：** * 检查事务阶段配置 *验证ApplicationEventPublisher是否正确自动连接 * 确保事务在事件发布之前提交 **卡夫卡连接问题：** * 验证引导服务器配置 * 检查 Kafka 的网络连接 * 确保正确的序列化配置 **事件处理失败：** * 检查事件处理程序中的循环依赖关系 * 验证交易边界 * 监控事件处理中的异常 ### 调试技巧 * 为 Spring 事件启用调试日志记录：`logging.level.org.springframework.context=DEBUG` * 使用相关 ID 来跟踪跨服务的事件 * 监控执行器端点中的事件处理指标 ## 限制和警告 * 使用 `@TransactionalEventListener` 发布的事件仅在事务提交后触发；确保这符合您的一致性要求。 * 避免在事件中发布大型对象，因为这可能会导致内存压力和序列化问题。 * 请谨慎对待异步事件处理程序，因为它们在单独的线程中执行，可能会导致并发问题。 * Kafka消费者必须通过实现幂等处理来处理重复消息。 * 分布式系统中不保证事件顺序；将处理程序设计为与订单无关。 * 切勿在主事务线程上运行的事件侦听器中执行阻塞操作。 * 监控事件处理积压，因为它们可以指示系统容量问题。 ## 示例 ### 输入：整体订单处理（反模式） @服务公共类 OrderService { @事务性公共订单 processOrder(OrderRequest 请求) { 订单 order = orderRepository.save(request); inventoryService.reserve(order.getItems()); paymentService.charge(order.getPayment()); ShippingService.schedule(订单); emailService.sendConfirmation(订单); 退货订单； } } ### 输出：事件驱动的订单处理 @服务公共类 OrderService { 私有最终 OrderRepository orderRepository; 私有最终ApplicationEventPublisher eventPublisher； @事务性公共订单 processOrder(OrderRequest 请求) { 订单 order = Order.create(request); orderRepository.save(order); // 事务提交后发布事件 eventPublisher.publishEvent(new OrderCreatedEvent( order.getId(), order.getItems(), order.getPayment() ））；退货订单； } } @组件公共类 OrderEventHandler { @TransactionalEventListener(阶段 = TransactionPhase.AFTER_COMMIT) 公共无效handleOrderCreated（OrderCreatedEvent事件）{ // 这些在订单保存后异步执行 inventoryService.reserve(event.getItems()); paymentService.charge(event.getPayment()); } } ### 输入：同步外部服务调用 @服务公共类通知服务{ 公共无效sendOrderNotification（订单顺序）{ emailClient.send(订单); // 阻塞调用 } } ### 输出：异步事件驱动通知公共类 OrderCreatedEvent 扩展 DomainEvent { 私有最终 OrderId orderId; 私人最终字符串客户电子邮件；私有最终 BigDecimal 总数； // 构造函数和 getter } @组件公共类NotificationEventHandler { @KafkaListener(主题 = "订单事件") 公共无效handleOrderEvent（OrderCreatedEventDto事件）{ // 异步处理，不阻塞订单流 emailService.sendOrderConfirmation（事件）； } } ### 输入：无法追踪的事件 eventPublisher.publishEvent(new ProductCreatedEvent(productId, name)); ### 输出：具有完全可追溯性的事件公共类 ProductCreatedEvent 扩展 DomainEvent { 私有最终EventId eventId = EventId.generate(); 私有最终 ProductId 产品Id；私有最终字符串名称；私人最终即时发生= Instant.now(); 私有最终 CorrelationId correlationId = CorrelationId.generate(); // 包括分布式跟踪的元数据公共地图<字符串，字符串> getMetadata（）{ 返回Map.of( "eventId", eventId.toString(), “correlationId”，correlationId.toString()， “时间戳”，发生At.toString() ）； } }

Spring Boot 事件驱动模式

包含技能