事件驱动架构核心
事件驱动不是「用消息队列」，而是一种松耦合的通信模式：
命令（Command） 事件（Event）
→ 意图明确 → 事实陈述
→ 点对点 → 广播/多订阅
→ 需要响应 → 无响应期望
→ 如：CreateOrder → 如：OrderCreated
eventd v2 内部使用事件驱动解耦模块：
OrderController (WebFlux)
→ OrderCreated Event
→ InventoryService（扣库存）
→ NotificationService（发通知）
→ AuditService（记日志）
Spring 事件（ApplicationEventPublisher）
Spring 的 ApplicationEventPublisher 可以发布事件，也可以在监听器里调用响应式服务，但发布和监听本身是同步的，不是响应式流：
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
@Component
public class OrderEventHandler {
private final ApplicationEventPublisher publisher;
public OrderEventHandler(ApplicationEventPublisher publisher) {
this.publisher = publisher;
}
public Mono<Order> createOrder(OrderRequest request) {
return validate(request)
.flatMap(this::save)
.doOnNext(saved -> publisher.publishEvent(new OrderCreatedEvent(
saved.id(), saved.symbol(), saved.quantity(), Instant.now())));
}
}
// 监听事件：用 @Async 放到独立线程，避免阻塞 Netty 事件循环
@Component
public class InventoryHandler {
@Async
@EventListener
public void onOrderCreated(OrderCreatedEvent event) {
deductInventory(event)
.doOnError(err -> log.error("Inventory deduction failed", err))
.subscribe();
}
}
publishEvent 是同步调用，@EventListener 会在发布线程上执行；如果监听器里有阻塞操作，需要 @Async 或把工作投递到线程池。@Async 还需要在配置类上 @EnableAsync，否则注解不会生效。监听器内部调用 .subscribe() 属于 fire-and-forget，异常和背压都不会回到发布方——这适合「通知、审计」这类可丢事件，不适合必须完成的业务动作。需要可靠执行时，把动作放进发布方的主链（flatMap 里 then 掉），或者交给下面的 Sinks 总线和单一订阅者。
响应式事件总线
Spring 事件是同步单进程的。更通用的方案是自定义响应式事件总线：
public interface DomainEvent {
String getType();
}
import reactor.core.publisher.Sinks;
import reactor.core.publisher.Flux;
import java.util.Objects;
@Component
public class ReactiveEventBus {
private final Sinks.Many<DomainEvent> sink =
Sinks.many().multicast().onBackpressureBuffer();
public Mono<Void> publish(DomainEvent event) {
// Mono.defer 让发送发生在订阅时，而不是组装时
return Mono.defer(() -> {
Sinks.EmitResult result = sink.tryEmitNext(event);
return result.isFailure()
? Mono.error(new IllegalStateException("Event publish failed: " + result))
: Mono.empty();
});
}
public Flux<DomainEvent> subscribe(String eventType) {
return sink.asFlux()
.filter(e -> Objects.equals(e.getType(), eventType));
}
}
multicast().onBackpressureBuffer() 默认只有 256 条共享缓冲，一个慢订阅者拖慢缓冲后，tryEmitNext 会持续返回 FAIL_OVERFLOW，直到该订阅者把积压消费掉；需要更大缓冲时显式传 onBackpressureBuffer(size)，并继续检查返回值。
它还有一层容易误读的预热语义：没有任何订阅者时 tryEmitNext 返回 OK，事件进入缓冲而不是被丢弃，第一个订阅者能收到订阅之前缓冲的事件；缓冲满了才会 FAIL_OVERFLOW。但它仍是热流：第一个订阅者之后的消费者收不到历史，进程崩溃时缓冲里的事件也会一起丢。换成 directAllOrNothing() / directBestEffort() 这类 direct 变体，零订阅者才会直接返回 FAIL_ZERO_SUBSCRIBER。所以不能把这里的 publish 当成可靠投递，集成事件要先落库或写 Kafka，再广播给进程内消费者。
使用（OrderCreatedEvent 的字段以本节下方的 record 定义为准）：
// 发布
bus.publish(new OrderCreatedEvent(
order.id(), order.symbol(), order.quantity(), Instant.now()))
.subscribe();
// 订阅
bus.subscribe("OrderCreated")
.flatMap(event -> process(event), 4) // 并发度 4
.subscribe();
领域事件 vs 集成事件
| 类型 | 范围 | 消费者 | 序列化 | 版本 |
|---|---|---|---|---|
| 领域事件 | 单个服务内部 | 进程内 | 不需要 | 随代码变更 |
| 集成事件 | 服务之间 | 跨进程 | JSON/Avro/Protobuf | 需兼容 |
eventd v2 的领域事件：
public record OrderCreatedEvent(
Long orderId,
String symbol,
int quantity,
Instant occurredOn
) implements DomainEvent {
public String getType() { return "OrderCreated"; }
}
集成事件（发送到外部系统）：
public record OrderCreatedIntegrationEvent(
String eventId, // 全局唯一
String eventType, // "order.created.v1"
Instant timestamp,
OrderPayload payload
) {}
public record OrderPayload(Long orderId, String symbol, int quantity) {}
集成事件需要版本号（v1、v2），支持消费者平滑升级。
事件网关
事件网关负责把内部领域事件转为外部集成事件：
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import reactor.core.Disposable;
import reactor.util.retry.Retry;
import java.time.Duration;
@Component
public class EventGateway {
private final ReactiveEventBus bus;
private final WebClient externalWebhook;
private Disposable subscription;
@PostConstruct
public void start() {
subscription = bus.subscribe("OrderCreated")
.map(this::toIntegrationEvent)
.flatMap(event -> externalWebhook.post()
.uri("/webhooks/order-created") // WebClient 必须显式给 URI，即使配了 baseUrl
.bodyValue(event)
.retrieve()
.toBodilessEntity()
.retryWhen(Retry.backoff(3, Duration.ofMillis(200)))
.onErrorResume(err -> {
log.error("Webhook failed, eventId={}", event.eventId(), err);
return Mono.empty(); // 转入死信/告警后丢弃，避免整条流终止
}), 4)
.subscribe();
}
@PreDestroy
public void stop() {
subscription.dispose();
}
private OrderCreatedIntegrationEvent toIntegrationEvent(DomainEvent event) {
OrderCreatedEvent e = (OrderCreatedEvent) event;
return new OrderCreatedIntegrationEvent(
UUID.randomUUID().toString(),
"order.created.v1",
Instant.now(),
new OrderPayload(e.orderId(), e.symbol(), e.quantity())
);
}
}
retrieve() 会把 4xx 和 5xx 都转成 WebClientResponseException，上面的 retryWhen 没有过滤，400/401 也会被重试三次。生产实现应只重试 5xx、超时和连接错误，并且确认下游接口对重复投递幂等；否则重试只会放大无效请求。
与外部系统的衔接
eventd v2 的事件流最终进入 Kafka（卷三处理）：
eventd v2 (内部事件总线)
→ OrderCreated / InventoryUpdated / PaymentProcessed
→ 事件网关
→ HTTP Webhook（实时通知下游）
→ Kafka Producer（异步持久化，卷三消费）
Kafka 衔接点在卷三的「Kafka Streams 与 RS 的衔接」章节展开。
事件序列化版本兼容
JSON 版本兼容策略：
{
"eventType": "order.created.v1",
"payload": { "orderId": 123, "symbol": "BTC" }
}
{
"eventType": "order.created.v2",
"payload": { "orderId": 123, "symbol": "BTC", "leverage": 1 }
}
v2 新增了 leverage 字段，v1 消费者忽略未知字段即可。Spring Boot 自动配置的 Jackson 默认关闭 FAIL_ON_UNKNOWN_PROPERTIES，所以消费方直接反序列化不会因新字段失败；如果消费方是手写 ObjectMapper 或其他语言，要显式确认这一行为。
升级 eventType 也有兼容成本：把 order.created.v1 直接改成 v2，按旧值路由的消费者会一条都收不到。上线期间通常要并行发布 v1/v2，等所有消费者迁移完成后再停掉 v1。
规则：
- 新增字段：消费者忽略未知字段
- 删除字段：发空值或默认值
- 修改字段：用新版本号，
eventType区分
小结
| 概念 | 实现 | 注意 |
|---|---|---|
| 内部事件 | ApplicationEventPublisher | @EventListener 同步，阻塞操作要 @Async |
| 响应式总线 | Sinks.Many + 检查 EmitResult | multicast 支持多订阅 |
| 领域事件 | 进程内，无序列化 | 随代码变更 |
| 集成事件 | JSON + 版本号 | 消费者需兼容旧版本 |
| 事件网关 | 内部 → 外部转换 | 错误处理 + 重试 |
| 版本兼容 | 新增字段忽略、删除字段默认 | eventType 区分版本 |
下一节 12.2 Saga 编排 ：事务补偿、最终一致性、Saga 状态机与响应式实现。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。