Spring Cloud Stream 消息驱动

怼烎@ 2024-03-22 22:06 140阅读 0赞

Stream为什么被引入

常见MQ(消息中间件)：

  • ActiveMQ
  • RabbitMQ
  • RocketMQ
  • Kafka

有没有一种新的技术诞生，让我们不再关注具体MQ的细节，我们只需要用一种适配绑定的方式，自动的给我们在各种MQ内切换。

Cloud Stream是什么？屏蔽底层消息中间件的差异，降低切换成本，统一消息的编程模型。

Stream是什么及Binder介绍

在这里插入图片描述

Stream的设计思想

标准MQ

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

为什么用Cloud Stream？

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

Stream凭什么可以统一底层差异

在这里插入图片描述

Binder

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

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

在这里插入图片描述

Stream编码常用注解简介

Spring Cloud Stream标准流程套路

在这里插入图片描述

在这里插入图片描述

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

编码API和常用注解

在这里插入图片描述

案例说明

  1. 准备RabbitMQ环境

  2. 工程中新建三个子模块

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

Stream消息驱动之生产者

1. 新建Module：cloud-stream-rabbitmq-provider8801
2. pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.     <parent>
  7.         <groupId>com.atguigu.springcloud</groupId>
  8.         <artifactId>cloud2020</artifactId>
  9.         <version>1.0-SNAPSHOT</version>
  10.     </parent>
  12.     <artifactId>cloud-stream-rabbitmq-provider8801</artifactId>
  14.     <dependencies>
  15.         <dependency>
  16.             <groupId>org.springframework.cloud</groupId>
  17.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  18.         </dependency>
  20.         <dependency>
  21.             <groupId>org.springframework.cloud</groupId>
  22.             <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
  23.         </dependency>
  25.         <dependency>
  26.             <groupId>org.springframework.boot</groupId>
  27.             <artifactId>spring-boot-starter-web</artifactId>
  28.         </dependency>
  30.         <dependency>
  31.             <groupId>org.springframework.boot</groupId>
  32.             <artifactId>spring-boot-starter-actuator</artifactId>
  33.         </dependency>
  35.         <dependency>
  36.             <groupId>org.springframework.boot</groupId>
  37.             <artifactId>spring-boot-devtools</artifactId>
  38.             <scope>runtime</scope>
  39.             <optional>true</optional>
  40.         </dependency>
  42.         <dependency>
  43.             <groupId>org.projectlombok</groupId>
  44.             <artifactId>lombok</artifactId>
  45.             <optional>true</optional>
  46.         </dependency>
  48.         <dependency>
  49.             <groupId>org.springframework.boot</groupId>
  50.             <artifactId>spring-boot-starter-test</artifactId>
  51.             <scope>test</scope>
  52.         </dependency>
  53.     </dependencies>
  55. </project>

3. application.yml

  1. server:
  2.   port: 8801
  4. spring:
  5.   application:
  6.     name: cloud-stream-provider
  7.   cloud:
  8.     stream:
  9.       binders: #此处配置要绑定的rabbitmq的服务信息
  10.         defaultRabbit: #定义的名称，用于binding整合
  11.           type: rabbit #消息组件类型
  12.           environment: #设置rabbitmq的相关环境配置
  13.             spring:
  14.               rabbitmq:
  15.                 host: localhost
  16.                 port: 5672
  17.                 username: guest
  18.                 password: guest
  19.       bindings: #服务的整合处理
  20.         output: #这个名字是一个通道的名称
  21.           destination: studyExchange #表示要使用的exchange的名称定义
  22.           content-type: application/json
  23.           binder: defaultRabbit #设置要绑定的消息服务的具体设置
  25. eureka:
  26.   client:
  27.     #表示是否注册到服务中心，默认是true, false表示不向服务中心注册自己
  28.     register-with-eureka: true
  29.     #是否丛服务端抓取已有的注册信息，默认为true,单节点无所谓，集群必需设置为true,才能使用ribbon使用负载均衡，false表示自己端就是服务中心，我的职责是维护实例，并不需要检索服务
  30.     fetchRegistry: true
  31.     service-url:
  32.       #设置与Eureka Server交互的地址查询服务和注册服务都需要依赖这个地址
  33.       defaultZone: http://localhost:7001/eureka/ #单机
  34.   #      defaultZone: http://eureka7001.com:7001/eureka/,http://eureka7002.com:7002/eureka/ #集群
  35.   instance:
  36.     instance-id: send-8801.com
  37.     prefer-ip-address: true #访问路径可以显示IP
  38.     #eureka客户端向发送心跳的等待时间上限，单位为秒(默认是30s)
  39.     lease-renewal-interval-in-seconds: 2
  40.     #eureka收到最后一次心跳等待时间上限，单位为秒(默认是90s),超过时间将踢除
  41.     lease-expiration-duration-in-seconds: 5

4. 主启动类StreamMQMain8801

  1. package com.atguigu.springcloud;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  6. @SpringBootApplication
  7. public class StreamMQMain8801 {
  9.     public static void main(String[] args) {
  11.         SpringApplication.run(StreamMQMain8801.class, args);
  12.     }
  13. }

5. 业务类
5.1 发送消息接口 service.IMessageProvider

  1. package com.atguigu.springcloud.service;
  3. public interface IMessageProvider {
  5.     public String sending();
  6. }

5.2 发送消息接口实现类

  1. package com.atguigu.springcloud.service.impl;
  3. import com.atguigu.springcloud.service.IMessageProvider;
  4. import org.springframework.cloud.stream.annotation.EnableBinding;
  5. import org.springframework.cloud.stream.messaging.Source;
  6. import org.springframework.messaging.MessageChannel;
  7. import org.springframework.messaging.support.MessageBuilder;
  9. import javax.annotation.Resource;
  10. import java.util.UUID;
  12. @EnableBinding(Source.class)  // 定义消息的推送管道
  13. public class MessageProviderImpl implements IMessageProvider {
  15.     @Resource
  16.     private MessageChannel output;  // 消息发送管道
  18.     @Override
  19.     public String sending() {
  21.         String serial = UUID.randomUUID().toString();
  22.         output.send(MessageBuilder.withPayload(serial).build());
  23.         System.out.println("**** serial: " + serial);
  24.         return null;
  25.     }
  26. }

5.3 controller.SendMessageController 类

  1. package com.atguigu.springcloud.controller;
  3. import com.atguigu.springcloud.service.IMessageProvider;
  4. import org.springframework.web.bind.annotation.GetMapping;
  5. import org.springframework.web.bind.annotation.RestController;
  7. import javax.annotation.Resource;
  9. @RestController
  10. public class ReceiveMessageListenerController {
  12.     @Resource
  13.     private IMessageProvider messageProvider;
  15.     @GetMapping("/sendMessage")
  16.     public String sendMessage() {
  18.         return messageProvider.sending();
  19.     }
  20. }

6. 测试

  • 启动 7001eureka

  • 启动 RabpitMq

    • rabbitmq-plugins enable rabbitmq_management
    • http://localhost:15672/
  • 启动 8801

  • 访问 - http://localhost:8801/sendMessage

    • 后台将打印serial: UUID字符串
      在这里插入图片描述

Stream消息驱动之消费者

1. 新建Module：cloud-stream-rabbitmq-consumer8802
2. pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.     <parent>
  7.         <groupId>com.atguigu.springcloud</groupId>
  8.         <artifactId>cloud2020</artifactId>
  9.         <version>1.0-SNAPSHOT</version>
  10.     </parent>
  12.     <artifactId>cloud-stream-rabbitmq-consumer8802</artifactId>
  14.     <dependencies>
  15.         <dependency>
  16.             <groupId>org.springframework.boot</groupId>
  17.             <artifactId>spring-boot-starter-web</artifactId>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.cloud</groupId>
  21.             <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
  22.         </dependency>
  23.         <dependency>
  24.             <groupId>org.springframework.cloud</groupId>
  25.             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.springframework.boot</groupId>
  29.             <artifactId>spring-boot-starter-actuator</artifactId>
  30.         </dependency>
  31.         <!--基础配置-->
  32.         <dependency>
  33.             <groupId>org.springframework.boot</groupId>
  34.             <artifactId>spring-boot-devtools</artifactId>
  35.             <scope>runtime</scope>
  36.             <optional>true</optional>
  37.         </dependency>
  38.         <dependency>
  39.             <groupId>org.projectlombok</groupId>
  40.             <artifactId>lombok</artifactId>
  41.             <optional>true</optional>
  42.         </dependency>
  43.         <dependency>
  44.             <groupId>org.springframework.boot</groupId>
  45.             <artifactId>spring-boot-starter-test</artifactId>
  46.             <scope>test</scope>
  47.         </dependency>
  48.     </dependencies>
  49. </project>

3. application.yml

  1. server:
  2.   port: 8802
  4. spring:
  5.   application:
  6.     name: cloud-stream-consumer
  7.   cloud:
  8.     stream:
  9.       binders: # 在此处配置要绑定的rabbitmq的服务信息；
  10.         defaultRabbit: # 表示定义的名称，用于于binding整合
  11.           type: rabbit # 消息组件类型
  12.           environment: # 设置rabbitmq的相关的环境配置
  13.             spring:
  14.               rabbitmq:
  15.                 host: localhost
  16.                 port: 5672
  17.                 username: guest
  18.                 password: guest
  19.       bindings: # 服务的整合处理
  20.         input: # 这个名字是一个通道的名称
  21.           destination: studyExchange # 表示要使用的Exchange名称定义
  22.           content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
  23.           binder: defaultRabbit # 设置要绑定的消息服务的具体设置
  25. eureka:
  26.   client: # 客户端进行Eureka注册的配置
  27.     service-url:
  28.       defaultZone: http://localhost:7001/eureka
  29.   instance:
  30.     lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
  31.     lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
  32.     instance-id: receive-8802.com  # 在信息列表时显示主机名称
  33.     prefer-ip-address: true     # 访问的路径变为IP地址

4. 主启动类StreamMQMain8802

  1. package com.atguigu.springcloud;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  6. @SpringBootApplication
  7. public class StreamMQMain8802 {
  9.     public static void main(String[] args) {
  11.         SpringApplication.run(StreamMQMain8802.class, args);
  12.     }
  13. }

5. 业务类 controller.ReceiveMessageListenerController

  1. package com.atguigu.springcloud;
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  6. @SpringBootApplication
  7. public class StreamMQMain8802 {
  9.     public static void main(String[] args) {
  11.         SpringApplication.run(StreamMQMain8802.class, args);
  12.     }
  13. }

6. 测试

  • 启动EurekaMain7001
  • 启动StreamMQMain8801
  • 启动StreamMQMain8802
  • 8801发送消息 运行地址： http://localhost:8801/sendMessage，8802接收消息
    在这里插入图片描述
    在这里插入图片描述

Stream之消息重复消费

1. 依照8802，克隆出来一份运行8803 - cloud-stream-rabbitmq-consumer8803。
2. 启动

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

3. 运行后有两个问题
在这里插入图片描述
4. 重复消费
在这里插入图片描述
在这里插入图片描述
5. 生产实际案例
在这里插入图片描述
在这里插入图片描述

Stream之group解决消息重复消费

在这里插入图片描述

原理

在这里插入图片描述
在这里插入图片描述
8802修改YML

  1. server:
  2.   port: 8802
  4. spring:
  5.   application:
  6.     name: cloud-stream-consumer
  7.   cloud:
  8.     stream:
  9.       binders: # 在此处配置要绑定的rabbitmq的服务信息；
  10.         defaultRabbit: # 表示定义的名称，用于于binding整合
  11.           type: rabbit # 消息组件类型
  12.           environment: # 设置rabbitmq的相关的环境配置
  13.             spring:
  14.               rabbitmq:
  15.                 host: localhost
  16.                 port: 5672
  17.                 username: guest
  18.                 password: guest
  19.       bindings: # 服务的整合处理
  20.         input: # 这个名字是一个通道的名称
  21.           destination: studyExchange # 表示要使用的Exchange名称定义
  22.           content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
  23.           binder: defaultRabbit # 设置要绑定的消息服务的具体设置
  24.           group: atguiguA  # 添加分组名称
  26. eureka:
  27.   client: # 客户端进行Eureka注册的配置
  28.     service-url:
  29.       defaultZone: http://localhost:7001/eureka
  30.   instance:
  31.     lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
  32.     lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
  33.     instance-id: receive-8802.com  # 在信息列表时显示主机名称
  34.     prefer-ip-address: true     # 访问的路径变为IP地址

8803修改YML

  1. server:
  2.   port: 8803
  4. spring:
  5.   application:
  6.     name: cloud-stream-consumer
  7.   cloud:
  8.     stream:
  9.       binders: # 在此处配置要绑定的rabbitmq的服务信息；
  10.         defaultRabbit: # 表示定义的名称，用于于binding整合
  11.           type: rabbit # 消息组件类型
  12.           environment: # 设置rabbitmq的相关的环境配置
  13.             spring:
  14.               rabbitmq:
  15.                 host: localhost
  16.                 port: 5672
  17.                 username: guest
  18.                 password: guest
  19.       bindings: # 服务的整合处理
  20.         input: # 这个名字是一个通道的名称
  21.           destination: studyExchange # 表示要使用的Exchange名称定义
  22.           content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”
  23.           binder: defaultRabbit # 设置要绑定的消息服务的具体设置
  24.           group: atguiguB  # 添加分组名称
  26. eureka:
  27.   client: # 客户端进行Eureka注册的配置
  28.     service-url:
  29.       defaultZone: http://localhost:7001/eureka
  30.   instance:
  31.     lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
  32.     lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
  33.     instance-id: receive-8803.com  # 在信息列表时显示主机名称
  34.     prefer-ip-address: true     # 访问的路径变为IP地址

在这里插入图片描述

在 RabbitMQ 页面可以看到分的组名
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
8802/8803都变成相同组，group两个相同，为group: atguiguA
在这里插入图片描述
这样通过分组就解决了重复消费的问题

Stream之消息持久化

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

总结： Spring Cloud Stream 添加分组 group 属性就可以避免重复消费以及实现持久化（消息丢失）

发表评论

表情：
评论列表 （有 0 条评论，140人围观）

还没有评论，来说两句吧...

相关阅读