Spring Cloud Stream 消息驱动微服务 相关图形

1. 微服务之间传递消息
   1. 使用消息中间件，在服务与服务之间传递消息 ，RabbitMQ/Kafaka

1. 消息中间件和服务之间耦合，换中间件，变动大/修改多         

1. 解耦
   1. 开发人员无感知地使用消息中间件
      1. 即可以做到代码层面对消息中间件的无感知
      2. 甚至于动态的切换中间件(rabbitmq切换为kafka)，实现轻松在不同消息中间件间切换的目的

Applicatoin core：微服务应用程序 + spring cloud stream框架。

inputs：输入，consumer消费者，从消息服务器中，消费消息。

outputs：输出，provider生产者，生产消息，把消息推送到消息服务器中。

消息服务器：限制rabbitmq、kafaka。

Binder：绑定器，包装了消息服务器。

             微服务应用 + spring cloud stream 《》Binder 《》消息服务器

1. 我们只需关注如何与Binder交互即可，而无需关注与具体消息中间件的交互。 
   1. 应用程序通过input（相当于消费者consumer）、output（相当于生产者producer）来与Spring Cloud Stream中Binder交互。

      1. Binder负责与消息中间件交互

1. Spring Cloud Stream 是一个构建消息驱动微服务的框架。
   1. Spring Cloud Stream对消息中间件的进一步封装
      1. ​​​​​​​​​​​​​​微服务在对接消息服务器时，按照统一的接口进行开发，屏蔽了底层的消息服务器的差异，实现底层消息服务器的自由切换​​​​​​​
         1. 有点类似于Hibernate，底层即可以连接mysql也可以连接oracle，但上层的Hibernate API却是一致的。即，如果使用了Hibernate，那么底层的数据库用什么都可以（可以使用mysql，不爽了，可以切换成oracle，dao层的代码不需要改变）。

1. 第一步：创建SpringBoot应用34-springcloud-service-stream；
2. 第二步：添加依赖

   
   
       org.springframework.cloud
       spring-cloud-starter-stream-rabbit
   

3. 第三步：配置文件

   #对接具体的消息中间件
   spring.cloud.stream.binders.rabbitmq.type=rabbit
   spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.host=192.168.10.128
   spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.port=5672
   spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.username=guest
   spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.password=guest
   spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.virtual-host=/
   
   #消息生产者
   #其中utput是一个key，这个名字是一个通道的名称，在代码中会用到
   #destination表示要使用的Exchange名称定义
   spring.cloud.stream.bindings.output.destination=spring.cloud.stream.exchange
   #设置要绑定的消息服务的binder
   spring.cloud.stream.bindings.output.binder=rabbitmq

4. 第四步：消息发送

   @EnableBinding(Source.class)
   public class MessageSender {
       @Autowired
       private MessageChannel output; // 消息的发送管道
   
       public void publish(String msg) {
           output.send(MessageBuilder.withPayload(msg).build());
           System.out.println("消息发送：<" + msg + "> 完成，时间：" + new Date());
       }
   }

5. 第五步：配置消息消费者

   #配置消息消费者
   #指定交换机
   spring.cloud.stream.bindings.input.destination=spring.cloud.stream.exchange
   #设置要绑定的消息服务的binder
   spring.cloud.stream.bindings.input.binder=rabbitmq

6. 第六步：消息接收

   @EnableBinding(Sink.class)
   public class MessageReceiver {
   
       @StreamListener(Sink.INPUT)
       public void input(Message message) {
           System.out.println("消息接收：<" + message.getPayload()  + "> 完成，时间：" + new Date());
       }
   }

5 只官配play：自定义消息通道（名称）

1. 第一步：接口

   public interface MessageSource {
   
       //channel名称
       String OUTPUT = "myOutput";
   
       @Output(OUTPUT)
       MessageChannel output();
   }

2. 第二步：接口

   public interface MessageSink {
   
       String INPUT = "myInput";
   
       @Input(Sink.INPUT)
       SubscribableChannel input();
   }

3. 配置消息消费者

   #配置消息消费者
   #指定交换机
   spring.cloud.stream.bindings.myInput.destination=spring.cloud.stream.exchange
   #设置要绑定的消息服务的binder
   spring.cloud.stream.bindings.myInput.binder=rabbitmq

4. 配置消息生产者

   #消息生产者
   #其中utput是一个key，这个名字是一个通道的名称，在代码中会用到
   #destination表示要使用的Exchange名称定义
   spring.cloud.stream.bindings.myOutput.destination=spring.cloud.stream.exchange
   #设置要绑定的消息服务的binder
   spring.cloud.stream.bindings.myOutput.binder=rabbitmq

6 只官配play：分组与持久化

1. 临时消息、不会被永久保留、可能会造成消息的丢失。

2. 分组的第一个作用是消息可以持久化（持久化消息）：

   1. 在Spring Cloud Stream 中在消费者端如果将队列设置为持久化队列，则队列名称会变为为destination.group，此时消费端的微服务宕机或重启，该队列信息依然会被保留在 RabbitMQ中，后续依然可以进行消费；

   2. 需要在消息的消费端配置一个分组，有分组就表示该消息可以进行持久化；

   3. #指定分组，可以进行消息的持久化 applies to consumers only
      spring.cloud.stream.bindings.myInput.group=rabbitmq-group