Appearance
Stream 详解：消息驱动
Spring Cloud Stream 是一个用于构建消息驱动微服务的框架。它的核心价值就一句话：让你用同一套代码操作 RabbitMQ、Kafka、RocketMQ，无需关心底层是哪个 MQ。
一、Stream 是必须的吗？
不是必须的。 你可以直接用 RabbitMQ 的原生 SDK、Kafka 的 KafkaTemplate、RocketMQ 的 RocketMQTemplate 来实现消息通信。Stream 只是一个"适配层"。
| 方式 | 代码示例 | 优缺点 |
|---|---|---|
| 直接用 MQ SDK | rabbitTemplate.convertAndSend(...) | 灵活，但绑定了特定 MQ |
| 用 Stream | @Bean Supplier<T> + 配置 | 换了 MQ 只需改配置，代码零改动 |
直接使用 MQ SDK：
┌──────────┐ ┌──────────┐
│ 服务 A │ rabbitTemplate.send() │ 服务 B │
│ (RabbitMQ)│ ──────────────────────→ │ (RabbitMQ)│
└──────────┘ └──────────┘
问题：哪天想换 Kafka，所有 rabbitTemplate 代码都要改
使用 Stream：
┌──────────┐ ┌──────────┐
│ 服务 A │ Supplier.send() │ 服务 B │
│ (Stream) │ ──────────────────────→ │ (Stream) │
└────┬─────┘ └────┬─────┘
│ │
┌────▼───────────────────────────────────▼─────┐
│ Binder 适配层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ RabbitMQ │ │ Kafka │ │ RocketMQ │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────┘
换 MQ：只改一个配置，业务代码一行不动结论：Stream 不是必须的，但如果你不确定未来用哪个 MQ，或者想降低 MQ 切换成本，用 Stream 是最佳选择。
二、为什么需要 Stream？
用一个比喻来理解：
没有 Stream：
你同时精通 中文、英文、法语、德语
跟不同的人说话要切换不同语言，很累
有 Stream：
你只需要说中文
Stream 自动帮你翻译成英文、法语、德语
你只管表达，不管翻译三个核心价值：
| 价值 | 说明 |
|---|---|
| 屏蔽 MQ 差异 | 一套代码，换 MQ 只改配置，业务代码零改动 |
| 统一编程模型 | 只学 Supplier/Function/Consumer 三个概念，不用学各 MQ 的 API |
| Spring 生态集成 | 天然支持 Spring Boot 配置、监控、Actuator 健康检查 |
三、快速上手
3.1 依赖
xml
<!-- RabbitMQ 方案 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- 或者 Kafka 方案 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>3.2 发送消息（生产者）
java
@Configuration
public class OrderProducer {
@Bean
public Supplier<OrderMessage> orderCreated() {
// Stream 会自动轮询这个 Supplier，拿到消息就发送
return () -> {
OrderMessage msg = orderQueue.poll(); // 从自己的队列取消息
return msg != null ? msg : null; // 返回 null 表示没有消息
};
}
}更实用的方式——用 StreamBridge 主动发送：
java
@RestController
public class OrderController {
@Autowired
private StreamBridge streamBridge;
@PostMapping("/orders")
public void createOrder(@RequestBody OrderDTO dto) {
// 业务逻辑...
// 发送消息：streamBridge.send("绑定名", 消息体)
streamBridge.send("orderCreated-out-0", new OrderMessage(dto.getId()));
}
}3.3 接收消息（消费者）
java
@Configuration
public class OrderConsumer {
@Bean
public Consumer<OrderMessage> orderCreated() {
return msg -> {
log.info("收到订单消息: {}", msg.getOrderId());
// 发短信、推送通知、更新统计等
};
}
}3.4 消息处理（Function）
java
@Bean
public Function<OrderMessage, PaymentMessage> orderToPayment() {
// 输入 OrderMessage → 处理 → 输出 PaymentMessage
// 可以串联多个 Function 形成处理链
return order -> {
PaymentMessage payment = new PaymentMessage();
payment.setOrderId(order.getOrderId());
payment.setAmount(order.getAmount());
return payment;
};
}3.5 配置
yaml
spring:
cloud:
stream:
# ===== 绑定配置 =====
bindings:
# 生产者：方法名-out-索引
orderCreated-out-0:
destination: order-topic # 对应 RabbitMQ Exchange / Kafka Topic
content-type: application/json
# 消费者：方法名-in-索引
orderCreated-in-0:
destination: order-topic
group: order-consumer-group # 消费者组（重要！）
content-type: application/json
# ===== 绑定器配置 =====
# RabbitMQ
rabbit:
bindings:
orderCreated-in-0:
consumer:
autoBindDlq: true # 自动创建死信队列
dlqTtl: 5000 # 死信队列消息 TTL
# Kafka
kafka:
bindings:
orderCreated-in-0:
consumer:
enableDlq: true命名规则总结：
<方法名>-<in/out>-<索引>
orderCreated-out-0 → 方法 orderCreated() 是生产者，输出通道 0
orderCreated-in-0 → 方法 orderCreated() 是消费者，输入通道 0
orderToPayment-in-0 → 方法 orderToPayment() 是 Function，输入通道 0
orderToPayment-out-0→ 方法 orderToPayment() 是 Function，输出通道 0四、核心原理
4.1 三大核心概念
┌──────────────────────────────────────────────────────────────────┐
│ Stream 三大核心概念 │
│ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Binder（绑定器） │ │
│ │ "翻译官"——把 Stream 的统一 API 翻译成各 MQ 的具体操作 │ │
│ │ 换 MQ = 换 Binder，业务代码不变 │ │
│ └────────────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Binding（绑定） │ │
│ │ "连接线"——把业务代码和 MQ 的 Topic/Queue 连起来 │ │
│ │ orderCreated-out-0 → order-topic（输出绑定） │ │
│ │ orderCreated-in-0 → order-topic（输入绑定） │ │
│ └────────────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Destination（目的地） │ │
│ │ "收件地址"——MQ 中的 Topic / Exchange / Queue │ │
│ │ RabbitMQ 中对应 Exchange，Kafka 中对应 Topic │ │
│ └────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘4.2 一条消息的完整旅程
┌─────────────────────────────────────────────────────────┐
│ 一条消息的完整旅程 │
└─────────────────────────────────────────────────────────┘
服务 A（生产者） 服务 B（消费者）
streamBridge.send( @Bean
"orderCreated-out-0", Consumer<OrderMessage> orderCreated() {
orderMessage) msg -> process(msg);
│ }
│ ▲
▼ │
┌──────────────────────┐ ┌──────────────────────┐
│ Binding │ │ Binding │
│ orderCreated-out-0 │ │ orderCreated-in-0 │
│ destination: │ │ destination: │
│ order-topic │ │ order-topic │
└──────────┬───────────┘ └──────────┬───────────┘
│ │
▼ │
┌──────────────────────┐ │
│ Binder (RabbitMQ) │ │
│ 发送到 RabbitMQ │ │
│ Exchange: order-topic│ │
└──────────┬───────────┘ │
│ │
▼ │
┌──────────────────────────────────────────────────┐
│ RabbitMQ Broker │
│ ┌────────────────────────────────────────────┐ │
│ │ Exchange: order-topic │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ Queue: order-topic.order-consumer-group │ │
│ │ (消费者组自动创建队列) │ │
│ └────────────────────────────────────────────┘ │
└──────────────────────┬───────────────────────────┘
│
▼
┌──────────────────────┐
│ Binder (RabbitMQ) │
│ 从 RabbitMQ 拉取消息 │
└──────────┬───────────┘
│
▼
orderCreated(msg)
消费消息4.3 框架如何自动发现 Bean？
这是 Stream 最巧妙的设计——不需要手动注册，框架自动扫描 Bean 并建立绑定。
Stream 启动时的自动发现流程：
1. Spring 容器启动，扫描所有 @Bean
2. 发现 Supplier<OrderMessage> orderCreated()
→ 这是一个生产者 → 寻找配置 bindings.orderCreated-out-0
→ 绑定到 destination: order-topic
3. 发现 Consumer<OrderMessage> orderCreated()
→ 这是一个消费者 → 寻找配置 bindings.orderCreated-in-0
→ 绑定到 destination: order-topic，group: order-consumer-group
4. 发现 Function<OrderMessage, PaymentMessage> orderToPayment()
→ 这是一个处理函数 → 寻找配置 bindings.orderToPayment-in-0 和 -out-0
→ 输入绑定到 order-topic，输出绑定到 payment-topic
完全约定优于配置：方法名是什么，绑定名就是什么！4.4 消费者组原理
这是最容易理解错的地方，用一张图讲清楚：
没有消费者组（group 不配置）：
order-topic
│
├──→ 实例1 消费 ← 第 1 份消息
└──→ 实例2 消费 ← 第 2 份消息（同一份消息被两个实例都消费了！）
问题：同一条消息被重复消费，这不是我们想要的
有消费者组（group = "sms-group"）：
order-topic
│
└──→ Queue: order-topic.sms-group
│
├──→ 实例1 消费消息1、3、5... ← 竞争关系
└──→ 实例2 消费消息2、4、6... ← 一条消息只被消费一次
正确：同一组内竞争消费，保证消息不重复
不同组：
order-topic
├──→ Queue: order-topic.sms-group
│ └──→ sms-service 实例们（发短信）
└──→ Queue: order-topic.email-group
└──→ email-service 实例们（发邮件）
不同组各自消费，互不影响（广播给不同业务）一句话：group 就是"谁和谁是一伙的"。同一 group 内竞争，不同 group 间广播。
五、高级特性
5.1 死信队列（DLQ）
yaml
spring:
cloud:
stream:
rabbit:
bindings:
orderCreated-in-0:
consumer:
autoBindDlq: true # 自动创建死信队列
dlqTtl: 5000 # 死信消息 5 秒后重新投递
maxAttempts: 3 # 最多重试 3 次正常流程：
消息 → orderCreated() 消费成功 → 确认 → 删除
异常流程：
消息 → orderCreated() 消费失败(异常)
→ 重试 1 → 失败
→ 重试 2 → 失败
→ 重试 3 → 失败
→ 进入死信队列(DLQ)
→ 人工处理或定时重投5.2 消息分区
java
// 保证同一个 orderId 的消息发到同一个分区（顺序消费）
@Bean
public Supplier<OrderMessage> orderCreated() {
return () -> { ... };
}yaml
spring:
cloud:
stream:
bindings:
orderCreated-out-0:
producer:
partition-key-expression: payload.orderId # 按 orderId 分区
partition-count: 4 # 4 个分区5.3 错误处理
java
@Bean
public Consumer<OrderMessage> orderCreated() {
return msg -> {
try {
process(msg);
} catch (Exception e) {
// 记录错误日志，发送告警
log.error("处理订单消息失败: {}", msg.getOrderId(), e);
throw e; // 抛出异常触发重试 + DLQ
}
};
}
// 全局错误处理
@Bean
public Consumer<ErrorMessage> errorHandler() {
return error -> {
log.error("Stream 全局错误: {}", error.getMessage());
// 发送钉钉/企业微信告警
};
}5.4 多输入/输出
java
// 一个服务可以消费多个 Topic，发送多个 Topic
@Configuration
public class MultiChannelConfig {
// 消费订单消息
@Bean
public Consumer<OrderMessage> orderCreated() {
return msg -> { ... };
}
// 消费支付消息
@Bean
public Consumer<PaymentMessage> paymentCompleted() {
return msg -> { ... };
}
// 发送通知消息
@Bean
public Supplier<NotifyMessage> notifyUser() {
return () -> { ... };
}
// 发送日志消息
@Bean
public Supplier<LogMessage> logRecord() {
return () -> { ... };
}
}六、Stream vs 直接使用 MQ SDK
| 维度 | Spring Cloud Stream | 直接使用 MQ SDK |
|---|---|---|
| 学习成本 | 低（只学 Supplier/Function/Consumer） | 高（需学各 MQ 特有 API） |
| 换 MQ 成本 | 极低（改 Binder 依赖 + 配置，代码不动） | 极高（重写所有发送/消费代码） |
| 代码量 | 少（声明式，Spring 自动处理大部分） | 多（手动管理连接、Channel、确认等） |
| 灵活性 | 中（抽象层限制，无法用 MQ 特有高级功能） | 高（可直接使用 MQ 全部特性） |
| 适用场景 | 标准的生产/消费，多 MQ 切换需求 | 需要 MQ 高级特性（事务消息、延迟消息等） |
| Spring 集成 | 天然集成（健康检查、监控、Actuator） | 需手动集成 |
选型建议：
简单生产消费 → Stream（代码少，维护简单）
需要 MQ 高级特性 → 原生 SDK（事务消息、顺序消息等）
不确定未来用哪个 MQ → Stream（换 MQ 成本最低）
团队熟悉 Spring 生态 → Stream（降低学习成本）七、面试要点
Q1：Spring Cloud Stream 的核心设计思想？
屏蔽 MQ 差异，统一编程模型。 通过 Binder 适配层，让开发者只需面向 Supplier/Function/Consumer 编程，无需关心底层是 RabbitMQ 还是 Kafka。核心概念：Binder（适配器）、Binding（连接线）、Destination（目的地）。
Q2：Stream 如何实现 MQ 切换？
换 Binder 依赖 + 改配置即可，业务代码零改动。例如从 RabbitMQ 切换到 Kafka，只需把 spring-cloud-starter-stream-rabbit 换成 spring-cloud-starter-stream-kafka，再修改 binder 配置。
Q3：消费者组的作用？
同一 group 内竞争消费（一条消息只被消费一次），不同 group 间广播消费（每个 group 都收到消息）。group 实现了消息的负载均衡和广播的灵活组合。
Q4：Stream 3.x 的函数式模型和旧版 @EnableBinding 有什么区别？
3.x 用 @Bean Supplier/Function/Consumer 替代了 @EnableBinding + @StreamListener。函数式模型更简洁，框架自动发现 Bean 并建立绑定，无需手动声明 Channel。
Q5：Stream 和 Bus 有什么区别？
| 维度 | Stream | Bus |
|---|---|---|
| 定位 | 业务消息处理（数据面） | 配置刷新广播（控制面） |
| 使用场景 | 下单、支付、通知等业务消息 | Config 配置批量刷新 |
| 关系 | Bus 底层就是 Stream 的封装 | 基于 Stream 构建 |
Q6：Stream 如何保证消息不丢失？
Stream 本身不保证，依赖于底层 MQ 的机制。通过配置 autoBindDlq（死信队列）+ maxAttempts（重试次数）+ 手动 ACK 模式，可以做到消费端消息不丢失。生产端可配合本地事务表实现最终一致性。