12.SpringCloud Stream消息驱动_messagechannel spring

1 消息驱动概述

1.1 是什么

• 屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型
• 官网
  https://spring.io/projects/spring-cloud-stream#overview
  https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html
  https://m.wang1314.com/doc/webapp/topic/20971999.html
  

1.2 设计思想

1.2.1 标准MQ

1.生产者/消费者之间靠消息媒介传递消息内容（Message）
1

2.消息必须走特定的通道
1

消息通道 MessageChannel

3.消息通道里的消息如何被消费呢，谁负责收发处理
1

消息通道MessageChannel的子接口SubscribableChannel，由MessageHandler消息处理器订阅

1.2.2 为什么用Cloud Stream

• stream凭什么可以统一底层差异
  
• Binder
  INPUT对应于消费者
  OUTPUT对应于生产者
  
  

1.2.3 Stream中的消息通信方式遵循了发布-订阅模式

Topic主题进行了广播

• 在RabbitMQ就是Exchange
• 在Kafka就是Topic

1.3 Spring Cloud Stream标准流程套路

• Binder
  很方便的连接中间件，屏蔽差异
• Channel
  通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置
• Source和Sink
  简单的可理解为参照对象是Spring Cloud Stream自身，从Stream发布消息就是输出，接受消息就是输入。

1.4 编码API和常用注解

2 案例说明

RabbitMQ环境已经OK

工程中新建三个子模块
1

cloud-stream-rabbitmq-provider8801作为生产者进行发消息模块
cloud-stream-rabbitmq-consumer8802作为消息接收模块
cloud-stream-rabbitmq-consumer8803作为消息接收模块

3 消息驱动之生产者

1.新建Module  cloud-stream-rabbitmq-provider8801
1

2.POM
1

3.YML
1

4.主启动类 StreamMQMain8801
1

5.业务类
1

6.测试
1

• 启动 7001 Eureka
• 启动 rabbitmq
• 启动 8801
• 访问 http://localhost:8801/sendMessage

4 消息驱动之消费者

1.新建Module  cloud-stream-rabbitmq-consumer8802
1

2.POM
1

3.YML
1

4.主启动类StreamMQMain8802
1

5.业务类
1

6.测试8801发送8802接收消息
1

http://localhost:8801/sendMessage

5 分组消费与持久化

1.依照8802，clone出来一份运行8803
1

cloud-stream-rabbitmq-consumer8803

2.启动
1

• RabbitMQ
• 7001 服务注册
• 8801 消息生产
• 8802 消息消费
• 8803 消息消费

3.运行后有两个问题
1

• 有重复消费问题
• 消息持久化问题

4.消费
1

目前是8802/8803同时都收到了，存在重复消费问题

• http://localhost:8801/sendMessage
• 如何解决（分组和持久化属性group）
• 生产实际案例

5.分组
1

1. 原理
   微服务应用放置于同一个group中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。
2. 8802/8803都变成不同组，group两个不同

   • group:atguiguA、atguiguB

   • 8802修改YML
     

   • 8803修改YML

   • 我们自己配置
     

   • 结论：还是重复消费

3. 8802/8803实现了轮询分组，每次只有一个消费者，8801模块的发的消息只能被8802或8803其中一个接收到，这样避免了重复消费。
4. 8802/8803都变成相同组，group两个相同
   • group：atguiguA
   • 8802修改YML
   • 8803修改YML
   • 结论：同一个组的多个微服务实例，每次只会有一个拿到。

6.持久化
1

• 通过上述，解决了重复消费问题，再看看持久化
• 停止8802/8803并去除8802的分组group:atguiguA(8803的分组group:atguiguA没有去掉)
• 8801先发送4条消息到rabbitmq
• 先启动8802，无分组属性配置，后台没有打出来消息
• 再启动8803，有分组属性配置，后台打出来了MQ上的消息