文章目录
前言
定义
Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于 Spring Boot 来创建独立的、可用于生产的 Spring 应用程序。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，并引入了发布-订阅、消费组、分区这三个核心概念。简单的说，Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration，实现了一套轻量级的消息驱动的微服务框架。抽象模型
我们都知道市面上有很多消息中间件，Sping Cloud Stream 为了可以集成各种各样的中间件，它抽象出了 Binder 的概念,每个消息中间件都需要有对应自己的 Binder。这样它就可以根据不同的 Binder 集成不同的中间件。下图的input和output是channel，Binder则是消息中间件和通道之间的桥梁。
绑定器
通过使用 Spring Cloud Stream，可以有效简化开发人员对消息中间件的使用复杂度，让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。但是目前 Spring Cloud Stream 只支持 RabbitMQ 和 Kafka 的自动化配置。
Spring Cloud Stream 提供了 Binder (负责与消息中间件进行交互)，我们则通过 inputs 或者 outputs 这样的消息通道与 Binder 进行交互。
Binder 绑定器是 Spring cloud Stream 中一个非常重要的概念，实现了应用程序和消息中间件之间的隔离，同时我们也可以通过应用程序实现，消息中间件之间的通信。在我们的项目的可以继承多种绑定器，我们可以根据不同特性的消息使用不同的消息中间件。Spring Cloud Stream 为我们实现了 RabbitMQ 和Kafka 的绑定器。如果你想使用其他的消息中间件需要自己去实现绑定器接口。
一、SpringCloud 集成 RocketMQ
1. pom 依赖
<!-- rocketmq -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency>
2. yml 配置
spring:
cloud:
stream:
function:
definition: producer1;consumer1 # 方法定义（用于定义发送者或消费者方法）
# 配置消息通道通用属性（适用于所有消息中间件）
bindings:
# 配置channel消息通道
consumer1-in-0:
destination: consumer_topic # topic消息主题
content-type: application/json # 内容格式
group: consumer-group # 消费者组
producer1-out-0:
destination: producer_topic # topic消息主题
content-type: application/json # 内容格式
rocketmq:
binder:
name-server: 127.0.0.1:9876 # rocketmq服务地址
vipChannelEnabled: true # 是否开启vip通道（兼容老版本使用。多监听一个端口用于接受处理消息，防止端口占用。）
# 配置消息通道独特属性（仅适用于rocketmq）
bindings:
# 配置channel消息通道（生产者：[functionName]-out-[index]，消费者：[functionName]-in-[index]）
producer1-out-0:
producer:
group: consumer-group
sync: true # 是否开启同步发送
consumer1-in-0:
consumer:
subscription: myTag # 消费tag
delayLevelWhenNextConsume: -1
suspendCurrentQueueTimeMillis: 99999999
broadcasting: false # 是否使用广播消费，默认为false使用集群消费
3. 操作实体
package com.demo.model;
import lombok.AllArgsConstructor;
import lombok.Data;
/**
* 消息model
*/
@Data
@AllArgsConstructor
public class MsgModel {
/**
* 消息id
*/
private String msgId;
/**
* 消息内容
*/
private String message;
}
4. 生产消息
4.1. 自动发送消息
通过 MessageBuilder 自动发送消息。
package com.demo;
import com.demo.model.MsgModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import java.util.function.Supplier;
/**
* 消息生产者类
*/
@Configuration
@Slf4j
public class MyProducer {
/**
* 消息生产者1
*/
@Bean
public Supplier<Message<MsgModel>> producer1() {
return () -> {
MsgModel msgModel = new MsgModel(System.currentTimeMillis(), "测试消息");
log.info("producer1发送消息：" + msgModel);
return MessageBuilder.withPayload(entity).build();
};
}
}
这种方式定义 suppelier 会 默认1000ms 发送一次记录。可以修改 spring.cloud.stream.poller.fixedDelay 设置延迟毫秒值。
4.2. 手动发送消息
通过 StreamBridge 手动发送消息。
package com.demo.controller;
import com.demo.model.MsgModel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 消息controller
*/
@RestController
@RequiredArgsConstructor
@RequestMapping("/msg")
@Slf4j
public class MsgController {
private final StreamBridge streamBridge;
/**
* 发送消息
*/
@GetMapping("/send")
public void sendMsg() {
MsgModel msgModel = new MsgModel(System.currentTimeMillis(), "测试消息");
log.info("producer1发送消息：" + msgModel);
streamBridge.send("producer1-out-0",
MessageBuilder.withPayload(entity).setHeader("MyHearder", "这是一个请求头").build());
}
}
5. 消费消息
package com.demo;
import com.demo.model.MsgModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import java.util.function.Consumer;
/**
* 消息消费者类
*/
@Configuration
@Slf4j
public class ReceiveMQ {
/**
* 消息消费者1
*/
@Bean
public Consumer<Message<MsgModel>> consumer1(){
return (message)->{
MessageHeaders headers = message.getHeaders();
MsgModel msgModel = message.getPayload();
log.info("consumer1接收消息，消息头：" + headers.get("MyHeader"));
log.info("consumer1接收消息，消息内容：" + msgModel);
};
}
}
二、配置解析
1. spring.cloud.stream.function.definition
进行生产者或消费者方法定义，在 rocketmq 初始时会加载这些方法以创建生产者或消费者列表。
不管是创建 Consumer 还是 Supplier 或者是 Function Stream 都会将其方法名称进行一个 topic 拆封和绑定。假设创建了一个 Consumer< String > myTopic 的方法，Stream 会将其 拆分成 In 和 out 两个通道：
- 输入通道（消费者）： [functionName]-in-[index]
consumer1-in-0 - 输出通道（生产者）： [functionName]-out-[index]
producer1-out-0
注意：这里的 functionName 需要和生产者或消费者方法名称以及 spring.cloud.stream.function.definition 下的名称保持一致。
