SpringCloud Stream--简单案例、分组消费与持久化

悠悠 2021-08-27 13:37 239阅读 0赞

#### 案例说明 ####

*  cloud-stream-rabbitmq-provider8801：作为生产者进行发消息模块
 *  cloud-stream-rabbitmq-provider8802：作为消费者接收消息模块
 *  cloud-stream-rabbitmq-provider8803：作为消费者接收消息模块

#### 生产者 ####

引入依赖：

<?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>cloud2020</artifactId>
            <groupId>pers.zhang.springcloud</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.boot</groupId>
                <artifactId>spring-boot-starter-actuator</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-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>

配置：application.yml

server:
      port: 8801
    
    spring:
      application:
        name: cloud-stream-provider
    
      cloud:
        stream:
          binders: #在此处配置要绑定的rabbitmq的服务信息
            defaultRabbit: #表示定义的命名，用于binding整合
              type: rabbit #消息组件类型
              environment: #设置rabbitmq的相关环境配置
                spring:
                  tabbitmq:
                    host: localhost
                    port: 5672
                    username: guest
                    password: guest
          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 #心跳时间间隔（默认30秒）
        lease-expiration-duration-in-seconds: 5 #超时间隔（默认90秒）
        instance-id: send-8801.com #在信息列表时显示主机名称
        prefer-ip-address: true #访问路径变为ip地址

消息接口及实现：

public interface IMessageProvider { 
        String send();
    }

@EnableBinding(Source.class) //定义消息的推送管道
    public class MessageProviderImpl implements IMessageProvider { 
    
        @Resource
        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:

@RestController
    public class SendMessageController { 
    
        @Resource
        IMessageProvider messageProvider;
    
        @GetMapping(value = "/sendMessage")
        public String sendMessage() { 
            return messageProvider.send();
        }
    }

启动类：

@SpringBootApplication
    public class StreamMQMain8801 { 
        public static void main(String[] args) { 
            SpringApplication.run(StreamMQMain8801.class, args);
        }
    }

测试：启动eureka，启动rabbitMQ，启动8801  
访问：`localhost:8801/sendMessage`  
![在这里插入图片描述][202003201254577.png]

后台打印：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

### 消费者 ###

引入依赖：

<?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>cloud2020</artifactId>
            <groupId>pers.zhang.springcloud</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>cloud-stream-rabbitmq-consumer8802</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>

配置：application.yml

server:
      port: 8802
    
    spring:
      application:
        name: cloud-stream-consumer
    
      cloud:
        stream:
          binders: #在此处配置要绑定的rabbitmq的服务信息
            defaultRabbit: #表示定义的命名，用于binding整合
              type: rabbit #消息组件类型
              environment: #设置rabbitmq的相关环境配置
                spring:
                  tabbitmq:
                    host: localhost
                    port: 5672
                    username: guest
                    password: guest
          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 #心跳时间间隔（默认30秒）
        lease-expiration-duration-in-seconds: 5 #超时间隔（默认90秒）
        instance-id: receive-8802.com #在信息列表时显示主机名称
        prefer-ip-address: true #访问路径变为ip地址

Listener：

@Component
    @EnableBinding(Sink.class)
    public class ReceiveMessageListenerController { 
    
        @Value("${server.port}")
        String serverPort;
    
        @StreamListener(Sink.INPUT)
        public void input(Message<String> message) { 
            System.out.println("消费者1号，------>接收到消息：" + message.getPayload() + "\tport:" + serverPort);
        }
    }

启动类：

@SpringBootApplication
    public class StreamMQMain8802 { 
        public static void main(String[] args) { 
            SpringApplication.run(StreamMQMain8802.class, args);
        }
    }

测试:启动eureka，启动RabbitMQ，启动8801生产者和8802消费者：

访问：`localhost:8801/sendMessage`  
![在这里插入图片描述][202003201317383.png]  
8802后台打印：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]

#### 分组消费与持久化 ####

依照8802，clone一个8803并启动：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]  
访问：`localhost:8801/sendMessage`  
![在这里插入图片描述][202003201327418.png]  
此时，8801和8802都接收到了消息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][20200320132825968.png]  
`存在重复消费问题！`

比如在如下场景中，订单系统我们做集群部署，都会从RabbitMQ中获取订单信息，那如果一个订单同时被两个服务获取到，那么就会造成数据错误，需要避免这种情况![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]  
这时就可以使用Stream中的消息分组来解决，在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的（重复消费）。同一个组内会发生竞争关系，只有其中一个可以消费。

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

**1. 使8802和8803变成不同的组：group: atguiguA、atguiguB**

8802修改yml：

bindings: #服务的整合处理
            input: #这个名字是一个通用的名称
              destination: studyExchange #表示要使用的Exchange名称定义
              content-type: application/json #设置消息类型，本次为json，文本则为 text/plain
              binder: defaultRabbit #设置要绑定的消息服务的具体设置
              group: atguiguA  #自定义分组

8803修改yml：

测试：  
![在这里插入图片描述][20200320185200476.png]  
发送消息：  
![在这里插入图片描述][20200320185249166.png]  
8802和8803均可以收到（不同组可以重复消费）：  
![在这里插入图片描述][20200320185325240.png]  
![在这里插入图片描述][20200320185335797.png]

**2. 使8802和8803实现轮询分组，每次只有一个消费者8801模块发送的消息只能被其中一个接收到这样就避免了重复消费：**

修改8803的group 为 atguiguA， 变成相同的组。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]

8801发送2条消息：  
![在这里插入图片描述][20200320185952839.png]  
8802接收到其中一条：  
![在这里插入图片描述][20200320190020212.png]  
8803接收到另一条：  
![在这里插入图片描述][20200320190046273.png]

#### 持久化 ####

1.  停止8802和8803微服务，并删除调8802的分组group:atguiguA
2.  发送4条消息到RabbitMQ  
    ![在这里插入图片描述][20200320191250527.png]
3.  先启动8802，无分组属性配置，后台没有打印消息  
    ![在这里插入图片描述][20200320191405649.png]
4.  再启动8803，有分组属性配置，打印全部4条消息  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]

`结论：` 配置了group属性的模块支持消息持久化

[202003201254577.png]: /images/20210813/ab0dc66923664efab9695f34f691a48f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/38ccdc90350c494280d558a0e272a585.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/2597372cbd3e4dc5a1f5fda59c2f363d.png
[202003201317383.png]: /images/20210813/3e6594ef88b74fc993c25b7575ec18f6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/8ef1c7ccc66a40e6ad94ddccd081695c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/5852b1b1c44e4ce1995cd635b7c0e492.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/9ca4d227af3f48a999155ae3ac939a16.png
[202003201327418.png]: /images/20210813/22d74fc045b34735a49c3da6e287bda9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/c723fbb9778e47059cff1a9935e9406b.png
[20200320132825968.png]: /images/20210813/f573bbe9fc82413c8917a0ea1d6e8704.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/2ed177e1ef5845319d3ad818769fbfc7.png
[20200320185200476.png]: /images/20210813/84a9d1fbf61346469216dab32c9e12e7.png
[20200320185249166.png]: /images/20210813/36e87dd16d8944809328277a18092200.png
[20200320185325240.png]: /images/20210813/cf1863dce03c428ebbd2a3c5bd4129a8.png
[20200320185335797.png]: /images/20210813/6cc87105403a4864a4c3612d2b007ec8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]: /images/20210813/da282b3f7f8d487c833aa4a42fe62003.png
[20200320185952839.png]: /images/20210813/d2ae4a9ea7c74e66a4523aed5ce10934.png
[20200320190020212.png]: /images/20210813/23ed9ea42b84495c8addffbbaaecf2d3.png
[20200320190046273.png]: /images/20210813/d2c3333296274c08a8cd79271980df85.png
[20200320191250527.png]: /images/20210813/b0eceddd7e584cb7ba4437993338451d.png
[20200320191405649.png]: /images/20210813/31925880c4a44a0c82ef03960db982b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]: /images/20210813/a126841e7cce4aeca85b74af559cc45b.png