博客学习参考视频

一、消息驱动概述

① 是什么

​ 一句话: 屏蔽底层消息中间件的差异,降低切换版本,统一消息的编程模型

image-20201021225204369

官网

  1. https://spring.io/projects/spring-cloud-stream#overview

    Spring Cloud Stream 是用于构建与共享 消息传递系统 连接的高度可伸缩的事件驱动微服务框架,该框架提供一个灵活的编程模型,它建立在 spring 的基础上,包括支持持久化的发布/订阅、消费以及分区 这三个核心概念

image-20201021230206112

  1. https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/
  2. Spring Cloud Stream 中 文 指 导 手 册 :https://m.wang1314.com/doc/webapp/topic/20971999.html

② 设计思想

标准 MQ

image-20201021230552252

  1. 生产者/消费者 之间靠消息媒介传递消息内容 – Message
  2. 消息必须走特定的通道 – 消息通道 MessageChannel
  3. 消息通道里的消息如何被消费呢,谁负责处理 – 消息通道 MessageChannel 的子接口 SubscribaleChannel, 由 MessageHandle 消息处理器订阅

为什么用 Cloud Stream

  1. 比方说我们用到的了 RabbitMQ 和 Kafka,由于这两个消息中间件的架构上的不同,像 RabbitMQ 有 exchange, kafka 有 Topic 和 Partitions 分区

image-20201021231225210

  1. 这些中间件的差异性导致我们实际项目开发中会造成一定困扰, 我们如果用了两个消息队列的其中一种, 后面的业务需求,我们想往另外一种消息队列进行迁移, 这时候无疑就是一个灾难性的, 一大堆东西都要中心推倒重新做, 因为他跟我们的系统耦合了, 这时候 springcloud Stream 给我们提供了一种解耦合的方式。
  2. stream 凭什么可以统一底层差异 。

image-20201021231415175

4.Binder

image-20201021231503099

image-20201021231646648

image-20201021231711183

  • INPUT 对应消费者
  • OUTPUT 对应于生产者

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

  1. Topic 主题进行广播
  • 在 RabbitMQ 就是 Exchange
  • 在 Kafka 中就是 Topic

③ Spring Cloud Stream 标准流程套路

image-20201021232144095

image-20201021232152904

  1. Binder:很方便的连接中间件,屏蔽差异

  2. Channel: 通道,是队列 Query 的一种抽象,在消息通讯系统中就是实现存储和转发的媒介,通过对 Channel 对队列进行配置

  3. Source 和 Sink: 简答的可理解为参照对象是 Spring Cloud Stream 自身,从 Stream 发布消息就是输出,接收消息就是输入

④ 编码 API 和常用注解

image-20201021232602501

二、案例说明

① RabbitaMQ 环境已经 OK

请参考这篇博客: SpringBoot 与消息

② 工程中新建三个子模块

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

三、消息驱动之生产者

新建 Module: cloud-stream-rabbitmq-provider8801

POM

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>clould</artifactId>
<groupId>com.oy</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

<artifactId>cloud-stream-rabbitmq-provider8801</artifactId>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</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>

</project>

YML

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
server:
port: 8801

spring:
application:
name: cloud-stream-provider
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息;
defaultRabbit: # 表示定义的名称,用于于binding整合
type: rabbit # 消息组件类型
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: 116.63.177.72
port: 5672
username: guest
password: guest
bindings: # 服务的整合处理
output: # 这个名字是一个通道的名称
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置

eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒)
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒)
instance-id: send-8801.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址

主启动类

1
2
3
4
5
6
@SpringBootApplication
public class StreamMQMain8801 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8801.class, args);
}
}

业务类

发消息接口:

1
2
3
4
public interface IMessageProvider {
public String send();
}

发送消息接口实现类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@EnableBinding(Source.class) // 定义消息的推送管道
public class MessionProvicerImpl implements IMessageProvider {

@Resource
private MessageChannel output; // 消息发送管道

@Override
public String send() {
String serial = UUID.randomUUID().toString();
output.send(MessageBuilder.withPayload(serial).build());
System.out.println("******serial:" + serial);
return null;
}
}

Controller

1
2
3
4
5
6
7
8
9
10
11
12
@RestController
public class SendMessageController {

@Resource
private IMessageProvider messageProvider;

@GetMapping(value = "/sendMessage")
public String sendMessage()
{
return messageProvider.send();
}
}

测试

注意:有个大坑,视频里的 application.yml 使用了 spring.cloud.stream.binders.defaultRabbit.environment.spring.rabbitmq.xx

来配置 rabbitmq 的环境如果你是用的其他服务器上的 rabbitmq,比如我使用的我自己的华为云服务器然后创建 docker 容器来运行

rabbitmq。按照视频中的配置方式的话,启动时会试图连接两次 rabbitmq 程序第一次试图连接访问的就是 application.yml 中配置

的地址,此时已经订阅成功了但是程序还会在之后进行第二次连接,此时访问的地址就是 localhost:5672,在我的环境中,我本地

没有 rabbitmq 环境,所以直接报 IOException 所以,如果是使用的自己的服务器来配置,则需要修改配置文件,将 rabbitmq 的配置

信息移动到 application.yml 中的 spring 节点下修改后的配置如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
server:
port: 8801

spring:
application:
name: cloud-stream-provider
rabbitmq:
host: 华为云ip
port: 5672
username: guest
password: guest
cloud:
stream:
binders:
defaultRabbit:
type: rabbit
bindings:
output:
destination: studyExchange
content-type: application/json
binder: defaultRabbit

eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒)
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒)
instance-id: send-8801.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址

四、消息驱动之消费者

新建 Module

​ cloud-stream-rabbitmq-consumer8802

POM

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>clould</artifactId>
<groupId>com.oy</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

<artifactId>cloud-stream-rabbitmq-consumer8802</artifactId>

<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!--
https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-eure
ka-server -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>com.oy</groupId>
<artifactId>cloud-api-commons</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<!--https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web
-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--
https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web
-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!--
https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-devtools -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId><optional>true</optional>
</dependency>
<!--
https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-test
-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

YML

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
server:
port: 8802

spring:
application:
name: cloud-stream-consumer
rabbitmq:
host: 华为云ip // localhost
port: 5672
username: guest
password: guest
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息;
defaultRabbit: # 表示定义的名称,用于于binding整合
type: rabbit # 消息组件类型

bindings: # 服务的整合处理
input: # 这个名字是一个通道的名称
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型,本次为对象json,如果是文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行 Eureka 注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是 30 秒)
lease-expiration-duration-in-seconds: 5 # 如果现在超过了 5 秒的间隔(默认是 90秒)
instance-id: receive-8802.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为 IP 地址

主启动

1
2
3
4
5
6
@SpringBootApplication
public class StreamMQMain8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8802.class, args);
}
}

业务类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package com.oy.springcloud.controller;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;

/**
* @Author OY
* @Date 2020/10/22
*/
@Component
@EnableBinding(Sink.class)
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);
}
}

测试 8801 发送 8802 接收消息:

http://localhost:8801/sendMessage

image-20201022222150351

image-20201022222211554

五、分组消费与持久化

① 配置 8803

依照 8802, clone 出来一份运行 8803

cloud-stream-rabbitmq-consumer8803

POM

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>clould</artifactId>
<groupId>com.oy</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>

<artifactId>cloud-stream-rabbitmq-consumer8803</artifactId>

<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</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>
</project>

YML

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
server:
port: 8803

spring:
application:
name: cloud-stream-consumer
rabbitmq:
host: 116.63.177.72
port: 5672
username: guest
password: guest
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息;
defaultRabbit: # 表示定义的名称,用于于binding整合
type: rabbit # 消息组件类型

bindings: # 服务的整合处理
input: # 这个名字是一个通道的名称
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型,本次为对象json,如果是文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行 Eureka 注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是 30 秒)
lease-expiration-duration-in-seconds: 5 # 如果现在超过了 5 秒的间隔(默认是 90秒)
instance-id: receive-8803.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为 IP 地址

主启动类

1
2
3
4
5
6
@SpringBootApplication
public class StreamMQMain8803 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8803.class, args);
}
}

Controller

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package com.oy.springcloud.controller;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

/**
* @Author OY
* @Date 2020/10/22
*/
@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {

@Value("${server.port}")
private String serverPort;

@StreamListener(Sink.INPUT)
public void input(Message<String> message){
System.out.println(" 消 费 者 2 号 , 接 受 : "+message.getPayload()+"\t port:"+serverPort);
}
}

② 启动

  • RabbitMQ
  • 7001: 服务注册
  • 8801: 消息生产
  • 8802: 消息消费
  • 8803: 消息消费
image-20201022232929071

③ 运行后两个问题

  1. 有重复消费问题
  2. 消息持久化问题 、

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

  1. http://localhost:8801/sendMessage

image-20201022232702874

image-20201022232716057

image-20201022232731143

  1. 如何解决分组和持久化属性 group
  2. 生产实际案例

image-20201022232842508

image-20201022232901129

⑤ 分组

原理

​ 微服务应用放置于同一个 group 中,就能保证消息只会被其中一个应用消费一次。不同的组是可以消费的。同一个组内发生的竞争的关系,只有一个可以消费。

8802/8803 都变成不同组, group 两个不同

  1. group: atguiguA、 atguiguB
  2. 8802 修改 YML
1
group: atguiguA

image-20201023105911727

  1. 8803 修改 YML
1
group: atguiguB

image-20201023110000699

  1. 我们自己配置

image-20201023110455388

​ 分布式微服务应用为了实现高可用和负载均衡,实际上都会户数多个实例,本例启动了两个消费微服务(8802/8803)多数情况,生产者发送消息给某个具体微服务只希望被消费一次,按照上面我们启动两个应用的例子,虽然它们同属一个应用,但是这个消息出现了被重复消费两次的情况。为了解决这个问题,在 Spring Cloud Stream 中提供了消费组的概念。

  1. 结论: 还是重复消费

8802/8803 实现了轮询分组, 每次只有一个消费者 8801 模块的发的消息只能被 8802 或 8803 其中一个接收到, 这样避免了重复

消费 。

8802/8803 都变成相同组, group 两个相同 。

  1. group: atguiguA
  2. 8802 修改 YML
1
group: atguiguA
  1. 8803 修改 YML
1
group: atguiguA

结论: 同一个组的多个微服务实例, 每次只会有一个拿到

image-20201023111418775

image-20201023111427639 image-20201023111445460

⑥ 持久化

通过上述, 解决了重复消费问题, 再看看持久化,停止 8802/8803 并去除掉 8802 的分组 group:atguiguA,8803 的分组

group:atguiguA 没有去掉。8801 先发送 4 条信息到 rabbitmq。

  • 先启动 8802, 无分组属性配置, 后台没有打出来消息
  • 先启动 8803, 有分组属性配置, 后台打出来了 MQ 上的消息

image-20201023111850171