Spring Cloud之-Spring Cloud Stream消息驱动- 15（个人笔记）

1、消息驱动概述 1-1、Spring Cloud Stream是什么

Spring Cloud Stream是一个构建消息驱动微服务的框架。应用程序通过Inpust消费者和Outputs对应于生产者与Spring Cloud中的Binder绑定器对象进行交互。首先我们需要配置绑定关系，也就是将具体的消息中间件和Stream进行绑定，后面使用Binder绑定器对象来操作消息中间件。此时，我们只需要了解Stream中的Binder绑定器对象相关的操作，即可完成消息中间件的交互。

Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。目前支持的消息中间件有：RabbitMQ和Kafka。Spring Cloud Stream目的在于屏蔽不同消息中间件的差异，统一消息的编程模型。

1-2、设计思想

对于普通的MQ，生产者和消费者之间依靠消息媒介传输信息，消息通过特定的通道进行传输，此时Spring Boot和消息中间件直接进行交互，因为不同消息中间件的设计不同，在实现细节上会有差异。通过定义Binder绑定器作为中间层，实现应用程序和消息中间件实现细节的隔离。Stream对消息中间件的进一步封装，可以做到代码层面对中间件的无感知，降低了学习成本。

Outputs对应消息生产者，向Binder发送消息，Binder将消息发给消息中间件，Inputs对应消息消费者，从Binder中取消息，Binder去消息中间件中获取。

Stream中的消息通信遵循发布-订阅模式，也就是通过Topic主题进行传播。

1-3、Spring Cloud Stream标准流程套路

1-4、编码API和常用注解

2、案例说明

为了演示Spring Cloud Stream，我们需要再创建3个子模块，分别是cloud-stream-rabbitmq-provider8801模块，cloud-stream-rabbitmq-consumer8802模块，cloud-stream-rabbitmq-consumer8803模块，另外，需要把RabbitMQ安装并启动。

3、消息驱动之生产者

创建cloud-stream-rabbitmq-provider8801模块，修改pom.xml，加入spring-cloud-starter-stream-rabbit的依赖。



    
    
        cloud2020
        com.king.springcloud
        1.0-SNAPSHOT
    

    4.0.0

    
    cloud-stream-rabbitmq-provider8801

    
        
        
            org.springframework.cloud
            spring-cloud-starter-stream-rabbit
        
        
        
            org.springframework.cloud
            spring-cloud-starter-netflix-eureka-client
        
        
        
            com.king.springcloud
            cloud-api-commons
            ${project.version}
        
        
        
            org.springframework.boot
            spring-boot-starter-web
        
        
        
            org.springframework.boot
            spring-boot-starter-actuator
        
        
        
            org.springframework.boot
            spring-boot-devtools
            runtime
            true
        
        
        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test

cloud-stream-rabbitmq-provider8801模块的application.yml文件：

# 配置服务端口号
server:
  port: 8801

# 配置应用信息
spring:
  application:
    name: cloud-stream-provider # 配置应用名称
  cloud:
    # 配置Spring cloud stream消息驱动
    stream:
      binders: # 此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示为binder绑定器对象定义一个名称，用于binding整合，随便取名字
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关环境配置
            spring:
              rabbitmq:  # 指定MQ的信息
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        output: # 一个通道的名称，不能修改，表示消息生产者发送
          destination: exchangeName # 目的地，表示RabbitMQ中要使用的Exchange交换机名称，给Exchange交换机起一个名字
          content-type: application/json # 设置消息类型，本次为json，文本要设置为“text/plain”
          binder: defaultRabbit # 表示binding对应哪个binder绑定器对象

# 配置eureka
eureka:
  client:
    service-url:
      # 设置与Eureka Server交互的地址，查询服务和注册服务都需要依赖这个地址
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30S）
    lease-expiration-duration-in-seconds: 5 # 如果超过5S间隔就注销节点（默认是90s）
    instance-id: provider-8801.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为IP地址

cloud-stream-rabbitmq-provider8801模块的主启动类：

package com.king.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StreamMQMain8801 {

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

cloud-stream-rabbitmq-provider8801模块的业务类：

package com.king.springcloud.service;


public interface IMessageProvider {
    
    public String send();
}

package com.king.springcloud.service.impl;

import com.king.springcloud.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import javax.annotation.Resource;
import java.util.UUID;


@EnableBinding(Source.class)
public class MessageProviderImpl implements IMessageProvider {
    
    @Resource
    private MessageChannel output;

    
    @Override
    public String send() {
        // 生成一个流水号
        String serial = UUID.randomUUID().toString();
        // 在消息管道弄一个绑定器，进行绑定消息，发送Message
        output.send(MessageBuilder.withPayload(serial).build());
        System.out.println("----------serial:" + serial);
        return null;
    }
}

cloud-stream-rabbitmq-provider8801模块的控制层：

package com.king.springcloud.controller;

import com.king.springcloud.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;


@RestController
public class SendMessageController {

    @Resource
    IMessageProvider messageProvider;

    
    @GetMapping("/sendMessage")
    public String sendMessage(){

        return messageProvider.send();
    }
}

启动Eureka7001，启动RabbitMQ服务，启动Provider8801模块测试。查看RabbitMQ管理后台，在Exchange标签下，可以看到我们定义的exchangeName。

浏览器访问http://localhost:8801/sendMessage发送消息，不断刷新页面发送请求，在控制台可以看到消息输出，在RabbitMQ管理后台的Overview可以看到消息发送的速率等信息。

4、消息驱动之消费者

新建cloud-stream-rabbitmq-consumer8802模块，修改pom.xml，依赖和cloud-stream-rabbitmq-provider8801一样。添加application.yml配置文件，注意这里通道名称改成input，之前cloud-stream-rabbitmq-provider8801使用的是output。

cloud-stream-rabbitmq-consumer8802模块的pom.xml文件：



    
    
        cloud2020
        com.king.springcloud
        1.0-SNAPSHOT
    

    4.0.0

    
    cloud-stream-rabbitmq-consumer8802

    
        
        
            org.springframework.cloud
            spring-cloud-starter-stream-rabbit
        
        
        
            org.springframework.cloud
            spring-cloud-starter-netflix-eureka-client
        
        
        
            com.king.springcloud
            cloud-api-commons
            ${project.version}
        
        
        
            org.springframework.boot
            spring-boot-starter-web
        
        
        
            org.springframework.boot
            spring-boot-starter-actuator
        
        
        
            org.springframework.boot
            spring-boot-devtools
            runtime
            true
        
        
        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test

cloud-stream-rabbitmq-consumer8802模块的application.yml文件：

# 配置服务端口号
server:
  port: 8802

# 配置应用信息
spring:
  application:
    name: cloud-stream-consumer # 配置应用名称
  cloud:
    # 配置Spring cloud stream消息驱动
    stream:
      binders: # 此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示为binder绑定器对象定义一个名称，用于binding整合，随便取名字
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关环境配置
            spring:
              rabbitmq:  # 指定MQ的信息
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 一个通道的名称，不能修改，表示消息消费者接收
          destination: exchangeName # 目的地，表示RabbitMQ中要使用的Exchange交换机名称，给Exchange交换机起一个名字
          content-type: application/json # 设置消息类型，本次为json，文本要设置为“text/plain”
          binder: defaultRabbit # 表示binding对应哪个binder绑定器对象

# 配置eureka
eureka:
  client:
    service-url:
      # 设置与Eureka Server交互的地址，查询服务和注册服务都需要依赖这个地址
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30S）
    lease-expiration-duration-in-seconds: 5 # 如果超过5S间隔就注销节点（默认是90s）
    instance-id: consumer-8802.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为IP地址

cloud-stream-rabbitmq-consumer8802模块的主启动类：

package com.king.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StreamMQMain8802 {

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

cloud-stream-rabbitmq-consumer8802模块的业务类：

package com.king.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;


@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {

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

    
    @StreamListener(Sink.INPUT)
    public void input(Message message) {
        // 因为cloud-stream-rabbitmq-provider8801模块发的是Message.withPayload()方法，这边就是接收Message.getPayload()方法
        System.out.println("ServerPort=" + serverPort + "消费者，消费消息：" + message.getPayload());
    }
}

启动RabbitMQ服务、cloud-eureka-server7001模块、cloud-stream-rabbitmq-provider8801模块、cloud-stream-rabbitmq-consumer8802模块，浏览器请求http://localhost:8801/sendMessage发送消息，在8802的控制台，可以看到消息接收的情况，测试通过。

5、分组消费与持久化

参照cloud-stream-rabbitmq-consumer8802模块，创建cloud-stream-rabbitmq-consumer8803模块并启动。

cloud-stream-rabbitmq-consumer8803模块的pom.xml文件：



    
    
        cloud2020
        com.king.springcloud
        1.0-SNAPSHOT
    

    4.0.0

    
    cloud-stream-rabbitmq-consumer8803

    
        
        
            org.springframework.cloud
            spring-cloud-starter-stream-rabbit
        
        
        
            org.springframework.cloud
            spring-cloud-starter-netflix-eureka-client
        
        
        
            com.king.springcloud
            cloud-api-commons
            ${project.version}
        
        
        
            org.springframework.boot
            spring-boot-starter-web
        
        
        
            org.springframework.boot
            spring-boot-starter-actuator
        
        
        
            org.springframework.boot
            spring-boot-devtools
            runtime
            true
        
        
        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test

cloud-stream-rabbitmq-consumer8803模块的application.yml文件：

# 配置服务端口号
server:
  port: 8803

# 配置应用信息
spring:
  application:
    name: cloud-stream-consumer # 配置应用名称
  cloud:
    # 配置Spring cloud stream消息驱动
    stream:
      binders: # 此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示为binder绑定器对象定义一个名称，用于binding整合，随便取名字
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关环境配置
            spring:
              rabbitmq:  # 指定MQ的信息
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 一个通道的名称，不能修改，表示消息消费者接收
          destination: exchangeName # 目的地，表示RabbitMQ中要使用的Exchange交换机名称，给Exchange交换机起一个名字
          content-type: application/json # 设置消息类型，本次为json，文本要设置为“text/plain”
          binder: defaultRabbit # 表示binding对应哪个binder绑定器对象

# 配置eureka
eureka:
  client:
    service-url:
      # 设置与Eureka Server交互的地址，查询服务和注册服务都需要依赖这个地址
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30S）
    lease-expiration-duration-in-seconds: 5 # 如果超过5S间隔就注销节点（默认是90s）
    instance-id: consumer-8803.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为IP地址

cloud-stream-rabbitmq-consumer8803模块的主启动类：

package com.king.springcloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StreamMQMain8803 {

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

cloud-stream-rabbitmq-consumer8803模块的业务类：

package com.king.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;


@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {

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

    
    @StreamListener(Sink.INPUT)
    public void input(Message message) {
        // 因为cloud-stream-rabbitmq-provider8801模块发的是Message.withPayload()方法，这边就是接收Message.getPayload()方法
        System.out.println("ServerPort=" + serverPort + "消费者，消费消息：" + message.getPayload());
    }
}

启动RabbitMQ服务、cloud-eureka-server7001服务注册中心模块、cloud-stream-rabbitmq-provider8801消息生产者模块、cloud-stream-rabbitmq-consumer8802消息消费者模块、cloud-stream-rabbitmq-consumer8803消息消费者模块，浏览器请求http://localhost:8801/sendMessage，结果8802和8803都消费了消息，存在重复消费和消息持久化的问题。

在Stream中，处于同一个group分组的多个消费者是竞争关系，一条消息只能消费一次，处于不同group分组的消费者是可以重复消费的。通过查看RabbitMQ的管理后台，点击exchange交换机标签，找到我们指定的exchange交换机，发现Bindings里面有两个组，所以出现了重复消费问题。

解决重复消费的办法也很简单，将不需要重复消费的消息指定同一分组即可。

修改两个consumer8802跟consumer8803的application.yml，在inputs结点下，加一个group属性，值为consumer随意取分组名。再次发送消息，查看8802和8803的控制台打印信息，可知同一个组的消费者每次只有一个可以消费消息，而且默认采用轮询的方式。

cloud-stream-rabbitmq-consumer8802模块的application.yml文件：

# 配置服务端口号
server:
  port: 8802

# 配置应用信息
spring:
  application:
    name: cloud-stream-consumer # 配置应用名称
  cloud:
    # 配置Spring cloud stream消息驱动
    stream:
      binders: # 此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示为binder绑定器对象定义一个名称，用于binding整合，随便取名字
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关环境配置
            spring:
              rabbitmq:  # 指定MQ的信息
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 一个通道的名称，不能修改，表示消息消费者接收
          destination: exchangeName # 目的地，表示RabbitMQ中要使用的Exchange交换机名称，给Exchange交换机起一个名字
          content-type: application/json # 设置消息类型，本次为json，文本要设置为“text/plain”
          binder: defaultRabbit # 表示binding对应哪个binder绑定器对象
          group: consumer # 消息分组并持久化配置，多个微服务消息在处于同一个group的多个消费者是竞争关系，一条消息只能消费一次，处于不同group分组的消费者是可以重复消费的看需求而定

# 配置eureka
eureka:
  client:
    service-url:
      # 设置与Eureka Server交互的地址，查询服务和注册服务都需要依赖这个地址
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30S）
    lease-expiration-duration-in-seconds: 5 # 如果超过5S间隔就注销节点（默认是90s）
    instance-id: consumer-8802.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为IP地址

cloud-stream-rabbitmq-consumer8803模块的application.yml文件：

# 配置服务端口号
server:
  port: 8803

# 配置应用信息
spring:
  application:
    name: cloud-stream-consumer # 配置应用名称
  cloud:
    # 配置Spring cloud stream消息驱动
    stream:
      binders: # 此处配置要绑定的rabbitmq的服务信息
        defaultRabbit: # 表示为binder绑定器对象定义一个名称，用于binding整合，随便取名字
          type: rabbit # 消息组件类型
          environment: # 设置rabbitmq的相关环境配置
            spring:
              rabbitmq:  # 指定MQ的信息
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 一个通道的名称，不能修改，表示消息消费者接收
          destination: exchangeName # 目的地，表示RabbitMQ中要使用的Exchange交换机名称，给Exchange交换机起一个名字
          content-type: application/json # 设置消息类型，本次为json，文本要设置为“text/plain”
          binder: defaultRabbit # 表示binding对应哪个binder绑定器对象
          group: consumer # 消息分组并持久化配置，多个微服务消息在处于同一个group的多个消费者是竞争关系，一条消息只能消费一次，处于不同group分组的消费者是可以重复消费的看需求而定

# 配置eureka
eureka:
  client:
    service-url:
      # 设置与Eureka Server交互的地址，查询服务和注册服务都需要依赖这个地址
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30S）
    lease-expiration-duration-in-seconds: 5 # 如果超过5S间隔就注销节点（默认是90s）
    instance-id: consumer-8803.com # 在信息列表时显示主机名称
    prefer-ip-address: true # 访问的路径变为IP地址

添加了group分组属性的消费者，在重启后，会自动获取未消费的消息，而没有添加group分组属性的消费者，在重启后，即使有未消费的消息，也无法消费了。比如，停掉8802和8803消费者，去掉8802的group属性，保留8803的group属性，生产者发送几条消息，重启8802模块，它并不能处理未消费的消息，重启8803模块，因为它指定了group分组，它可以拿到未消费的消息，继续消费，实现了消息的持久化。指定过group的消费者重启时候，在RabbitMQ的Exchange中，会继续保留group的名称，用于下次重启消费者时候，将消息再次发送给消费者，未指定group名称的消费者，每次启动，都会随机生成一个group放在Exchange的Bindings里面，当消费者重启，会把这个随机值从Bindings里移除，再加入一个新的随机值，所以说，未指定group分组的消费者，无法获取到之前未消费的消息，也就没法继续消费了。

Spring Cloud之-Spring Cloud Stream消息驱动- 15（个人笔记）

Java相关栏目本月热门文章