事件驱动架构：Spring Cloud Stream 与领域事件

事件驱动架构：Spring Cloud Stream 与领域事件 – 程序员的午后咖啡

各位看官，大家好！我是你们的老朋友，江湖人称“Bug终结者”的程序猿老王。今天咱们不聊996的悲惨故事，也不谈内卷的血雨腥风，咱们来一杯香醇的“事件驱动”咖啡，聊聊如何用 Spring Cloud Stream 玩转领域事件，让你的微服务架构变得优雅、高效、可扩展！

什么是事件驱动架构？别怕，不是玄学！

别被“事件驱动架构”这个高大上的名字吓到，其实它就像你每天早上听到的闹钟一样简单。闹钟响了，这是一个“事件”，你听到闹钟响了，这就是“事件驱动”，然后你起床了，这就是“事件处理”。

在软件世界里，事件驱动架构（EDA）就是一种构建应用程序的方式，应用程序中的组件通过异步地生成和消费事件进行通信。不再是服务 A 直接调用服务 B，而是服务 A 发出一个“事件”，告诉大家“我干了XX事情”，然后服务 B、服务 C、甚至服务 D，谁感兴趣谁来处理这个事件。

打个比方：

想象一下你经营一家咖啡店，以前客人点一杯咖啡，你需要亲自跑去后厨，告诉咖啡师“来一杯拿铁”。现在有了 EDA，你只需要喊一声“来一杯拿铁”，然后咖啡师听到后自己去做，收银员听到后自己算账，配送员听到后自己准备送餐。你只需要发出一个“点单”事件，其他人都根据自己的职责来处理。

EDA 的优点：

• 解耦： 服务之间不再直接依赖，修改一个服务不会影响其他服务。
• 可扩展： 可以轻松地添加新的服务来处理事件，提高系统的吞吐量。
• 弹性： 如果某个服务处理事件失败，不会影响其他服务。
• 实时性： 事件可以实时地传递和处理，适用于需要快速响应的场景。

EDA 的缺点：

• 复杂性： 需要考虑事件的格式、路由、持久化、错误处理等问题。
• 调试困难： 由于是异步通信，排查问题比较困难。
• 事务一致性： 需要考虑分布式事务的问题。

领域事件：事件驱动架构的灵魂

领域事件是指发生在领域模型内部，并且领域专家关心的事件。它代表了业务上的一个重要的状态变化。

继续用咖啡店的例子：

• 点单事件 (OrderCreated)： 客人下单了，订单状态变为“已创建”。
• 支付事件 (PaymentReceived)： 客人支付了，订单状态变为“已支付”。
• 咖啡制作完成事件 (CoffeeBrewed)： 咖啡师做好了咖啡，订单状态变为“制作完成”。
• 订单完成事件 (OrderCompleted)： 客人取走了咖啡，订单状态变为“已完成”。

领域事件的特点：

• 领域相关： 事件的名称和数据应该反映业务上的含义。
• 不可变： 事件发生后就不能修改，代表了一个已经发生的事实。
• 及时性： 事件应该在状态变化发生后立即发布。

领域事件的作用：

• 驱动领域模型状态变化： 领域事件可以触发领域模型的其他操作，比如更新订单状态、发送短信通知等。
• 解耦领域模型和其他模块： 领域事件可以将领域模型的变化通知给其他模块，而不需要直接依赖。
• 支持事件溯源： 可以通过记录所有的领域事件来还原领域模型的历史状态。

Spring Cloud Stream：事件驱动架构的瑞士军刀

Spring Cloud Stream 是一个构建事件驱动微服务应用程序的框架。它简化了与消息中间件（比如 RabbitMQ、Kafka）的集成，提供了一种统一的方式来发布和消费事件。

Spring Cloud Stream 的核心概念：

• Binder： 连接应用程序和消息中间件的桥梁。Spring Cloud Stream 提供了多种 Binder 实现，用于支持不同的消息中间件。
• Source： 用于发布事件的组件。
• Sink： 用于消费事件的组件。
• StreamListener： 用于监听事件的注解。
• MessageChannel： 用于发送和接收消息的通道。

Spring Cloud Stream 的优点：

• 简化集成： 屏蔽了底层消息中间件的细节，提供了一致的编程模型。
• 配置灵活： 可以通过配置文件来调整应用程序的行为。
• 易于测试： 提供了测试工具，可以方便地测试事件的发布和消费。

实战：用 Spring Cloud Stream 实现咖啡店的事件驱动架构

接下来，我们用 Spring Cloud Stream 来实现咖啡店的事件驱动架构。我们将使用 RabbitMQ 作为消息中间件。

1. 创建 Spring Boot 项目：

使用 Spring Initializr 创建一个 Spring Boot 项目，添加以下依赖：

• spring-cloud-starter-stream-rabbit
• spring-boot-starter-web
• lombok

2. 定义领域事件：

import lombok.Data;

import java.math.BigDecimal;

@Data
public class OrderCreatedEvent {
    private String orderId;
    private String customerId;
    private String coffeeType;
    private BigDecimal price;
}

@Data
public class PaymentReceivedEvent {
    private String orderId;
    private String paymentId;
    private BigDecimal amount;
}

@Data
public class CoffeeBrewedEvent {
    private String orderId;
    private String coffeeMakerId;
}

@Data
public class OrderCompletedEvent {
    private String orderId;
}

3. 创建 Source 组件：

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
@EnableBinding(Source.class)
public class OrderSource {

    private final Source source;

    public OrderSource(Source source) {
        this.source = source;
    }

    public void publishOrderCreatedEvent(OrderCreatedEvent event) {
        source.output().send(MessageBuilder.withPayload(event).build());
        System.out.println("发布 OrderCreatedEvent: " + event);
    }

    public void publishPaymentReceivedEvent(PaymentReceivedEvent event) {
        source.output().send(MessageBuilder.withPayload(event).build());
        System.out.println("发布 PaymentReceivedEvent: " + event);
    }

    public void publishCoffeeBrewedEvent(CoffeeBrewedEvent event) {
        source.output().send(MessageBuilder.withPayload(event).build());
        System.out.println("发布 CoffeeBrewedEvent: " + event);
    }

    public void publishOrderCompletedEvent(OrderCompletedEvent event) {
        source.output().send(MessageBuilder.withPayload(event).build());
        System.out.println("发布 OrderCompletedEvent: " + event);
    }
}

4. 创建 Sink 组件：

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;

@Component
@EnableBinding(Sink.class)
public class OrderSink {

    @StreamListener(Sink.INPUT)
    public void handleOrderCreatedEvent(OrderCreatedEvent event) {
        System.out.println("接收到 OrderCreatedEvent: " + event);
        // 处理订单创建事件，比如发送短信通知
    }

    @StreamListener(Sink.INPUT)
    public void handlePaymentReceivedEvent(PaymentReceivedEvent event) {
        System.out.println("接收到 PaymentReceivedEvent: " + event);
        // 处理支付事件，比如更新订单状态
    }

    @StreamListener(Sink.INPUT)
    public void handleCoffeeBrewedEvent(CoffeeBrewedEvent event) {
        System.out.println("接收到 CoffeeBrewedEvent: " + event);
        // 处理咖啡制作完成事件，比如通知配送员
    }

    @StreamListener(Sink.INPUT)
    public void handleOrderCompletedEvent(OrderCompletedEvent event) {
        System.out.println("接收到 OrderCompletedEvent: " + event);
        // 处理订单完成事件，比如记录销售数据
    }
}

5. 创建 Controller：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.math.BigDecimal;
import java.util.UUID;

@RestController
public class OrderController {

    @Autowired
    private OrderSource orderSource;

    @GetMapping("/order/{customerId}/{coffeeType}/{price}")
    public String createOrder(@PathVariable String customerId, @PathVariable String coffeeType, @PathVariable BigDecimal price) {
        String orderId = UUID.randomUUID().toString();

        OrderCreatedEvent orderCreatedEvent = new OrderCreatedEvent();
        orderCreatedEvent.setOrderId(orderId);
        orderCreatedEvent.setCustomerId(customerId);
        orderCreatedEvent.setCoffeeType(coffeeType);
        orderCreatedEvent.setPrice(price);

        orderSource.publishOrderCreatedEvent(orderCreatedEvent);

        PaymentReceivedEvent paymentReceivedEvent = new PaymentReceivedEvent();
        paymentReceivedEvent.setOrderId(orderId);
        paymentReceivedEvent.setPaymentId(UUID.randomUUID().toString());
        paymentReceivedEvent.setAmount(price);

        orderSource.publishPaymentReceivedEvent(paymentReceivedEvent);

        CoffeeBrewedEvent coffeeBrewedEvent = new CoffeeBrewedEvent();
        coffeeBrewedEvent.setOrderId(orderId);
        coffeeBrewedEvent.setCoffeeMakerId(UUID.randomUUID().toString());

        orderSource.publishCoffeeBrewedEvent(coffeeBrewedEvent);

        OrderCompletedEvent orderCompletedEvent = new OrderCompletedEvent();
        orderCompletedEvent.setOrderId(orderId);

        orderSource.publishOrderCompletedEvent(orderCompletedEvent);

        return "Order created with id: " + orderId;
    }
}

6. 配置 application.yml：

spring:
  cloud:
    stream:
      bindings:
        output:
          destination: orders
          contentType: application/json
        input:
          destination: orders
          group: coffee-shop
      rabbit:
        binder:
          auto-bind-dlq: true # 自动创建死信队列
          dlq-ttl: 60000 # 死信队列消息过期时间，单位毫秒

7. 运行项目：

启动 Spring Boot 项目，访问 http://localhost:8080/order/customer1/latte/30， 你会看到控制台输出：

发布 OrderCreatedEvent: OrderCreatedEvent(orderId=..., customerId=customer1, coffeeType=latte, price=30)
发布 PaymentReceivedEvent: PaymentReceivedEvent(orderId=..., paymentId=..., amount=30)
发布 CoffeeBrewedEvent: CoffeeBrewedEvent(orderId=..., coffeeMakerId=...)
发布 OrderCompletedEvent: OrderCompletedEvent(orderId=...)
接收到 OrderCreatedEvent: OrderCreatedEvent(orderId=..., customerId=customer1, coffeeType=latte, price=30)
接收到 PaymentReceivedEvent: PaymentReceivedEvent(orderId=..., paymentId=..., amount=30)
接收到 CoffeeBrewedEvent: CoffeeBrewedEvent(orderId=..., coffeeMakerId=...)
接收到 OrderCompletedEvent: OrderCompletedEvent(orderId=...)

代码解释：

• @EnableBinding(Source.class) 和 @EnableBinding(Sink.class) 注解分别开启了 Source 和 Sink 功能。
• Source.output() 方法获取了用于发布事件的 MessageChannel。
• Sink.INPUT 常量表示用于消费事件的 MessageChannel。
• @StreamListener(Sink.INPUT) 注解表示监听 Sink.INPUT 通道上的事件。
• spring.cloud.stream.bindings.output.destination 和 spring.cloud.stream.bindings.input.destination 配置了事件的交换机名称。
• spring.cloud.stream.bindings.input.group 配置了消费者组，同一个消费者组内的消费者只会消费一次事件。

进阶：错误处理、消息转换、分区

1. 错误处理：

在事件驱动架构中，错误处理非常重要。Spring Cloud Stream 提供了多种错误处理机制：

• 死信队列 (Dead Letter Queue)： 如果消费者处理事件失败，可以将事件发送到死信队列，方便后续分析和处理。
• 重试机制： 可以配置消费者自动重试处理失败的事件。

配置死信队列：

在 application.yml 中添加以下配置：

spring:
  cloud:
    stream:
      rabbit:
        bindings:
          input:
            consumer:
              auto-bind-dlq: true # 自动创建死信队列
              requeue-rejected: false # 是否将拒绝的消息重新放回队列
        binder:
          auto-bind-dlq: true # 自动创建死信队列
          dlq-ttl: 60000 # 死信队列消息过期时间，单位毫秒

配置重试机制：

spring:
  cloud:
    stream:
      bindings:
        input:
          consumer:
            max-attempts: 3 # 最大重试次数
            back-off-initial-interval: 1000 # 初始重试间隔，单位毫秒
            back-off-max-interval: 10000 # 最大重试间隔，单位毫秒
            back-off-multiplier: 2 # 重试间隔倍数

2. 消息转换：

Spring Cloud Stream 默认使用 Jackson 来序列化和反序列化消息。如果需要使用其他序列化方式，可以配置 contentType 属性。

spring:
  cloud:
    stream:
      bindings:
        output:
          contentType: application/json # 使用 JSON 格式
        input:
          contentType: application/json # 使用 JSON 格式

3. 分区：

如果需要提高系统的吞吐量，可以使用分区。分区可以将事件分散到多个消费者上，每个消费者只处理一部分事件。

配置分区：

spring:
  cloud:
    stream:
      bindings:
        output:
          producer:
            partition-key-expression: payload.orderId # 使用 orderId 作为分区键
            partition-count: 3 # 分区数量
        input:
          consumer:
            instance-count: 3 # 消费者实例数量
            instance-index: 0 # 消费者实例索引

代码解释：

• partition-key-expression 指定了用于计算分区键的表达式。
• partition-count 指定了分区的数量。
• instance-count 指定了消费者实例的数量。
• instance-index 指定了消费者实例的索引，每个消费者实例的索引必须不同。

总结：让事件驱动成为你的架构利器

通过本文，我们一起探索了事件驱动架构的魅力，并学习了如何使用 Spring Cloud Stream 来构建事件驱动的微服务应用程序。从简单的咖啡店点单系统，到复杂的错误处理、消息转换和分区，希望这些知识能够帮助你更好地理解和应用事件驱动架构。

记住，事件驱动架构不是银弹，它有自己的适用场景和局限性。在选择架构时，需要根据实际情况进行权衡。

最后，祝大家编程愉快，Bug 永远远离！下次再见！