spring-boot-event-driven-patterns 是做什么的？
为 Spring Boot 提供事件驱动架构 (EDA) 模式——包括创建领域事件、配置 ApplicationEvent 和 @TransactionalEventListener、设置 Kafka 生产者和消费者，以及实现用于可靠分布式消息传递的事务性发件箱模式。用于在 Spring Boot 中实现事件驱动系统、设置基于 Kafka 的异步消息传递、从 DDD 聚合中发布领域事件，或需要可靠事件发布的发件箱模式场景。
使用场景
- 在 Spring Boot 应用程序中实现领域事件
- 配置
@TransactionalEventListener以在事务成功提交后处理事件 - 集成 Kafka 作为事件代理
- 创建可靠的发件箱模式，确保事件不会丢失
- 在服务之间异步通信
配置
Kafka 依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
application.yml 配置示例
spring:
kafka:
bootstrap-servers: localhost:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
consumer:
group-id: event-group
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring.json.trusted.packages: "*"
基本用法
创建领域事件
public class OrderCreatedEvent implements DomainEvent {
private final String orderId;
private final LocalDateTime occurredAt;
// 构造函数
}
发布事务性事件
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleOrderCreated(OrderCreatedEvent event) {
// 事务提交后处理事件
}
配置 Kafka 生产者
@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
发件箱模式
事务性发件箱模式确保事件在数据库事务成功后才发布。它通过将事件存储在与业务数据相同的数据库中，并单独处理这些事件来工作，从而保证可靠性。
发件箱表结构
CREATE TABLE outbox_events (
id UUID PRIMARY KEY,
aggregate_type VARCHAR(255),
aggregate_id VARCHAR(255),
event_type VARCHAR(255),
payload JSONB,
created_at TIMESTAMP,
processed BOOLEAN DEFAULT false
);
调度器处理已提交事件
@Component
public class OutboxPoller {
@Scheduled(fixedDelay = 5000)
@Transactional
public void processOutbox() {
List<OutboxEvent> unprocessedEvents = outboxRepository.findAllByProcessedFalse();
for (OutboxEvent event : unprocessedEvents) {
kafkaTemplate.send(event.getTopic(), event.getPayload());
event.setProcessed(true);
outboxRepository.save(event);
}
}
}