事件驱动架构模式：从事件总线到Kafka Streams与CQRS

事件驱动架构，说白了就是让系统里的各个模块通过「事件」来沟通。我最早接触这个概念是在做一个电商订单系统的时候——订单创建后要通知库存、通知物流、通知财务，如果全用同步调用，一个环节挂了整个流程就卡死了。后来改成事件驱动，世界清净了。

事件驱动架构的核心三要素

事件驱动架构有三个核心角色：事件源、事件总线、事件处理器。咱们一个一个说。

事件源（Event Source）

事件源就是产生事件的地方。比如用户下单、支付成功、商品上架，这些都是事件源。我习惯把事件源设计成轻量级的——它只管发布事件，不关心谁消费。你想想看，如果事件源还要知道下游有多少消费者，那耦合度就太高了。

事件总线（Event Bus）

事件总线是事件的中转站。它负责接收事件源发布的事件，然后分发给感兴趣的处理器。我见过不少团队自己手写事件总线，说实话，小项目还行，一旦上了规模，还是用现成的中间件靠谱。

常见的实现方式有：

• 内存事件总线：Guava EventBus、Spring ApplicationEvent
• 消息队列：RabbitMQ、Kafka、RocketMQ
• 云原生事件总线：AWS EventBridge、Azure Event Grid

事件处理器（Event Handler）

事件处理器就是真正干活的。它订阅感兴趣的事件类型，收到事件后执行对应的业务逻辑。我曾经犯过一个错误——在事件处理器里做了太多事情，导致处理时间过长，拖慢了整个事件总线。后来我学乖了：每个处理器只做一件事，做完了就返回。

一个简单的Java事件驱动示例

先看一个最基础的实现，用Java的EventBus。这个例子能帮你理解事件驱动的核心流程。

// 定义事件
public class OrderCreatedEvent {
    private final String orderId;
    private final String userId;
    private final BigDecimal amount;
    
    public OrderCreatedEvent(String orderId, String userId, BigDecimal amount) {
        this.orderId = orderId;
        this.userId = userId;
        this.amount = amount;
    }
    // getters...
}

// 事件处理器
public class InventoryHandler {
    @Subscribe
    public void handleOrderCreated(OrderCreatedEvent event) {
        System.out.println("库存服务：扣减库存，订单ID=" + event.getOrderId());
        // 实际业务逻辑...
    }
}

public class NotificationHandler {
    @Subscribe
    public void handleOrderCreated(OrderCreatedEvent event) {
        System.out.println("通知服务：发送订单确认消息给用户=" + event.getUserId());
        // 实际业务逻辑...
    }
}

// 事件总线使用
public class OrderService {
    private final EventBus eventBus = new EventBus();
    
    public OrderService() {
        eventBus.register(new InventoryHandler());
        eventBus.register(new NotificationHandler());
    }
    
    public void createOrder(Order order) {
        // 业务逻辑...
        eventBus.post(new OrderCreatedEvent(order.getId(), 
            order.getUserId(), order.getAmount()));
    }
}

你看，OrderService只管发布事件，InventoryHandler和NotificationHandler各自处理自己的逻辑。它们之间没有任何直接依赖。这就是事件驱动的魅力。

Kafka Streams：流式事件处理

当事件量大了以后，内存事件总线就不够用了。这时候需要Kafka这样的分布式消息系统。Kafka Streams是Kafka自带的流处理库，它让你可以用类似写普通程序的方式处理事件流。

我参与过一个实时风控系统，每秒要处理上万笔交易事件。用Kafka Streams做事件处理，代码量不大，性能却很好。

// Kafka Streams 事件处理示例
public class OrderEventProcessor {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        StreamsBuilder builder = new StreamsBuilder();
        
        // 从Kafka主题读取事件流
        KStream<String, OrderEvent> orderStream = 
            builder.stream("order-events", 
                Consumed.with(Serdes.String(), new OrderEventSerde()));
        
        // 过滤、转换、聚合
        KStream<String, AlertEvent> highValueOrders = orderStream
            .filter((key, event) -> event.getAmount().compareTo(new BigDecimal("10000")) > 0)
            .mapValues(event -> new AlertEvent(event.getOrderId(), 
                "高价值订单", event.getAmount()));
        
        // 输出到另一个主题
        highValueOrders.to("high-value-alerts", 
            Produced.with(Serdes.String(), new AlertEventSerde()));
        
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

这里有个坑要注意：Kafka Streams的Serde（序列化/反序列化）必须自己实现，或者用Avro、Protobuf这类序列化框架。我刚开始用的时候忘了自定义Serde，结果跑起来全是乱码。

CQRS：命令查询职责分离

CQRS和事件驱动经常一起出现。CQRS的核心思想是：读操作和写操作走不同的路径。为什么这么做？因为读和写的关注点完全不同。

维度	写操作（Command）	读操作（Query）
数据模型	面向业务逻辑，通常用领域模型	面向展示，通常用DTO/视图模型
一致性要求	强一致性	最终一致性即可
扩展性	垂直扩展为主	可以水平扩展
缓存策略	一般不缓存	大量使用缓存

我做过一个报表系统，刚开始读写混在一起，查询慢得要命。后来拆成CQRS架构，写端用MySQL，读端用Elasticsearch，查询性能提升了10倍不止。

事件驱动 + CQRS 的经典架构

把事件驱动和CQRS结合起来，就形成了下面这个经典架构。我画了一张图帮你理解。

这张图展示了完整的流程：用户发送命令到命令处理器，命令处理器把事件写入事件存储，事件存储把事件推送到事件总线，事件总线分发给各个事件处理器，事件处理器更新读数据库。用户查询时直接走读数据库，不经过写路径。

避坑指南

做事件驱动架构，有几个坑我踩过，分享给你：

• 事件顺序问题：同一个订单的多个事件，如果顺序乱了，数据就错了。Kafka的同一个分区能保证顺序，但跨分区就不行了。我建议按业务ID做分区键。
• 幂等性设计：事件可能被重复消费。我曾经因为没做幂等，导致库存多扣了一次。解决方案是在事件处理器里加去重逻辑，或者用数据库的唯一约束。
• 事件版本管理：事件结构会变。我习惯在事件里加一个version字段，处理器根据版本做兼容处理。
• 死信队列：处理失败的事件不能丢。我一般会配置死信队列，把处理失败的事件存起来，人工介入处理。

总结一下

事件驱动架构的核心就三件事：事件源发布事件，事件总线分发事件，事件处理器消费事件。加上CQRS后，读写分离，各自优化。Kafka Streams让流式事件处理变得简单，适合大规模场景。

我个人的建议是：从小处着手，先在一个子系统中试用事件驱动，跑通了再推广。别一上来就搞全系统重构，那样风险太大。

---

公众号:蓝海资料掘金营，微信deep3321