简单搭建，没有技术含量，Demo可用
1、介绍
①产生原因
RabbitMQ、RocketMQ、Kafka、ActiveMQ
在一个项目中，可能存在多种不同的MQ，在不同的MQ中，切换维护开发都很麻烦。
如果你会RabbitMQ，不会Kafka，要换MQ，还要重头学？？
有没有技术，能够不再关注MQ细节，让我们使用一种适配绑定的版本，自动在MQ内切换：
有，SpringCloud Stream
②简介
SpringCloud Stream是一个构建消息驱动微服务的框架
官方文档：Spring Cloud Stream
应用程序通过inputs和outputs来和SpringCloud Stream的binder对象交互。
我们通过配置来binding（绑定），binder对象负责和消息中间件交互。
我们只需要和Spring Cloud Stream交互就能使用MQ了
就类似于JDBC能够操作Mysql，Orcal等不同数据库
通过Spring Integration来连接消息代理中间件，以实现消息事件驱动
SpringCloud Stream为一些中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区三个核心概念。
③支持MQ
④设计思路
在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接和MQ进行交互
通过定义绑定器作为中间层，我们能够完美实现应用程序和消息中间件细节之间的隔离。
通过向应用程序暴露统一的Channel通道，使应用程序不需要考虑各种不同的消息中间件实现。
通过定义绑定器Binder作为中间层，实现应用程序和消息中间件细节间的隔离
通过定义绑定器Binder作为中间层，实现应用程序和消息中间件之间的隔离
⑤注解简介
| 组成 | 说明 |
|---|---|
| Middleware | 中间件，只支持RabbitMQ和Kafka |
| Binder | Binder是应用和消息中间件之间的封装，通过Binder可以方便的连接中间件，可以动态改变消息类型（Kafka的topic，RabbitMq的exchange），可以通过配置文件实现 |
| @Input | 注解表示输入通道，通过输入通道收到的消息进入ApplicationCore |
| @Output | 注解标识输出通道，发布的消息通过该通道离开ApplicationCore |
| @StreamListener | 监听队列，用于消费者的队列的消息接收 |
| @EnableBinding | 指信道channel和exchange绑定在一起 |
2、QuickStart
①Provider搭建
新建cloud-stream-rabbitmq-provider8801，生产者
pom.xml：
<dependencies>
<!--stream rabbit -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!--eureka client-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--监控-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!--热部署-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
yml：
server:
port: 8801
spring:
application:
name: cloud-stream-provider
rabbitmq:
host: 101.43.244.40
port: 5672
username: guest
password: guest
cloud:
stream:
binders:
defaultRabbit:
type: rabbit
bindings: #服务的整合处理
output: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”
binder: defaultRabbit #设置要绑定的消息服务的具体设置（爆红不影响使用，位置没错）
eureka:
client:
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)
lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s
instance-id: send-8801.com #在信息列表时显示主机名称
prefer-ip-address: true #访问的路径变为IP地址
启动类：
@SpringBootApplication
public class StreamMQMain8801 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8801.class, args);
}
}
业务类：
新建service.IMessageProvider接口
public interface IMessageProvider {
public String send();
}
@EnableBinding(Source.class) //定义消息的推送管道（Source是spring的）
public class IMessageProviderImpl implements IMessageProvider {
@Resource
private MessageChannel output; //消息发送管道
@Override
public String send() {
String serial = UUID.randomUUID().toString();
output.send(MessageBuilder.withPayload(serial).build()); //MessageBuilder是spring的integration.support.MessageBuilder
System.out.println("*******serial: " + serial);
return null;
}
}
新建controller.SendMessageController
@RestController
public class SendMessageController {
@Resource
private IMessageProvider iMessageProvider;
@GetMapping("/sendMessage")
public String sendMessage(){
return iMessageProvider.send();
}
}
启动Eureka注册中心，Provider8801，查看MQ控制台面板，Exchange成功创建：
向Provider接口发送请求：
http://localhost:8801/sendMessage
控制台成功打印
查看RabbitMQ控制台，能够看到发送的消息：
②Consumer搭建
新建模块cloud-stream-rabbitmq-consumer8802
pom.xml：
<dependencies>
<!--stream rabbit -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!--eureka client-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--监控-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!--热部署-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
yml.xml：
server:
port: 8802
spring:
application:
name: cloud-stream-consumer
rabbitmq:
host: 101.43.244.40
port: 5672
username: guest
password: guest
cloud:
stream:
binders:
defaultRabbit:
type: rabbit
bindings: #服务的整合处理
input: #这个名字是一个通道的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”
binder: defaultRabbit #设置要绑定的消息服务的具体设置（爆红不影响使用，位置没错）
eureka:
client:
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)
lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s
instance-id: receive-8802.com #在信息列表时显示主机名称
prefer-ip-address: true #访问的路径变为IP地址
启动类：
@SpringBootApplication
public class StreamMQMain8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8802.class, args);
}
}
Controller：
@EnableBinding(Sink.class)
@Controller
public class ReceiveMessageListenerController {
@Value("${server.port}")
private String serverPort;
@StreamListener(Sink.INPUT) //监听
public void input(Message<String> message){
System.out.println("消费者1号------>收到的消息：" + message.getPayload() + "\t port：" + serverPort);
}
}
③测试
启动Eureka7001，Provider8801，Consumer8802
向8801发送请求：
http://localhost:8801/sendMessage
8802接收请求：
关于之前8801发送的消息，为什么8802没有接收到呢？
因为在8802启动之前，StudyExchange被创建出，但是没有binding队列，所以消息发送的Exchange就丢失了。
这些丢失的消息，在Consuemr启动，binding队列到Exchange后，不会被接收到
3、分组消费
①项目搭建
拷贝Consumer8802，新建Consumer8803子模块
启动：
②重复消费问题
如图，8801发送消息，8802和8803都可以对该消息进行消费
为什么？
因为8802和8803各自对Exchange创建了队列，并且绑定到了8801创建的Exchange上，所以Exchange会分发两条相同的消息到他们各自的队列上，他们也会通过各自的队列消费：
③解决：分组
同一个组的多个微服务实例，每次只会有一个拿到Provider产生的消息。而不同组的多个微服务，每个组都可以消费一次：
组？：
如下图所示，队列的Queue是由 Exchange的名词+
.+组名 组成的
将Consumer8802和Consumer8803设置为同一个组：
在它们的配置中添加group都设置为GroupA
测试：
8802为8803都为GroupA，此时消息只会被消费一次
