Spring Cloud Stream 消息通信⽅式及编程模型

站长

2023年05月23日 23:24 · 阅读数 57

前言

Spring Cloud Stream消息驱动组件是 Spring Cloud 中组件之一，通常用来帮助我们更快速的，更方便，更友好，优雅的方式去构建消息驱动微服务。

为什么不直接使用消息队列如ActiveMQ RabbitMQ RocketMQ Kafka等。

而是使用Spring Cloud Stream，它有什么优点，解决了什么样的问题？

Stream解决问题

首先对MQ中间件来说，有很多产品，其中包括ActiveMQ RabbitMQ RocketMQ Kafka等，MQ消息中间件⼴泛应⽤在应⽤解耦合、异步消息处理、流量削峰等场景中。

不同的MQ消息中间件内部机制包括使⽤⽅式都会有所不同，⽐如RabbitMQ中有Exchange（交换机/交换器）. kafka有Topic、Partition分区这些概念，MQ消息中间件之间的差异性不利于我们上层的开发应⽤。如果我们有需要从一个MQ切换到另一个MQ，那对应我们的程序代码也需要进行修改，此时MQ和程序耦合起来了。

此时，Stream就解决了这个问题，底层可以任意切换MQ，目前Stream仅支持RocketMQ 和 Kafka两款MQ产品。

Spring Cloud Stream进⾏了很好的上层抽象、屏蔽掉了底层具体MQ消息中间件的细节差异，就像Hibernate屏蔽掉了具体数据库（Mysql/Oracle⼀样）。

Stream重要概念

Spring Cloud Stream 是⼀个构建消息驱动微服务的框架。应⽤程序通过inputs（相当于消息消费者consumer）或者outputs（相当于消息⽣产者producer）来与Spring Cloud Stream中的binder对象交互，⽽Binder对象是⽤来屏蔽底层MQ细节的，它负责与具体的消息中间件交互。

Spring Cloud Stream 消息通信⽅式及编程模型

Application core：消费者逻辑代码，生产者逻辑代码 MiddleWare : 对应底层中间件MQ inputs: 可以用多个input 、关联消费者 输入相对于程序代码来说从Mq输入到消费者进行消费。 outputs: 可以用多个output、关联生产者 输出相对于应用程序来说，从程序输出到mq中。 通道（Binddings）: 通道用来跟Binder打交道 Binder: 绑定器对象，用来和底层MQ打交道

传统MQ模型与Stream消息驱动模型 Spring Cloud Stream 消息通信⽅式及编程模型

Binder绑定器

Binder绑定器是Spring Cloud Stream 中⾮常核⼼的概念，就是通过它来屏蔽底层不同MQ消息中间件的细节差异，当需要更换为其他消息中间件时，我们需要做的就是更换对应的Binder绑定器⽽不需要修改任何应⽤逻辑（Binder绑定器的实现是框架内置的，Spring Cloud Stream⽬前⽀持Rabbit、Kafka两种消息队列）

Stream消息通信⽅式

Stream中的消息通信⽅式遵循了发布—订阅模式。

在Spring Cloud Stream中的消息通信⽅式遵循了发布-订阅模式，当⼀条消息被投递到消息中间件之后，它会通过共享的 Topic 主题进⾏⼴播，消息消费者在订阅的主题中收到它并触发⾃身的业务逻辑处理。这⾥所提到的 Topic 主题是SpringCloud Stream中的⼀个抽象概念，⽤来代表发布共享消息给消费者的地⽅。在不同的消息中间件中， Topic 可能对应着不同的概念，⽐如：在RabbitMQ中的它对应了Exchange、在Kakfa中则对应了Kafka中的Topic。

Stream编程注解

Spring Cloud Stream 消息通信⽅式及编程模型

代码示例

首先创建三个工程，服务注册端就不创建了直接8761

server_pruduce_stream_server_7100 生产者
server_consumer_stream_server_7101 消费者
server_consumer_stream_server_7102 消费者

生产者

依赖

   <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>



        <!--Config 客户端依赖-->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-config-client</artifactId>
        </dependency>


        <!-- bus自动刷新 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bus-amqp</artifactId>
        </dependency>

配置输入，以及mq配置

#eureka server服务端口
server:
  port: 7100
spring:
  application:
    name: server-pruduce-stream-server-7100 # 应用名称，应用名称会在Eureka中作为服务名称

  cloud:
    inetutils:
      # 指定此客户端的ip
      default-ip-address: springcloud

# stream 配置
    stream:
      binders: # 绑定MQ服务信息（此处我们是RabbitMQ）
        testRabbitBinder: # 给Binder定义的名称，⽤于后⾯的关联
          type: rabbit # MQ类型，如果是Kafka的话，此处配置kafka
          environment:
            spring:
              rabbitmq:
                # MQ环境配置（⽤户名、密码等）
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 关联整合通道和binder对象
        output: # output是我们定义的通道名称，此处不能乱改
          destination: mqExchange # 要使⽤的Exchange名称（消息队列主题名称）
          content-type: text/plain # application/json # 消息类型设置，⽐如json
          binder: testRabbitBinder # 关联MQ服务


eureka:
  instance:
    hostname: springcloud  # 当前eureka实例的主机名
    ip-address: springcloud
  client:
    service-url:
      prefer-ip-address: true
      lease-renewal-interval-in-seconds: 30
      # 租约到期，服务时效时间，默认值90秒,服务超过90秒没有发⽣⼼跳，服务注册中心会将服务从列表移除
      lease-expiration-duration-in-seconds: 90
      # 配置客户端所交互的Eureka Server的地址（Eureka Server集群中每一个Server其实相对于其它Server来说都是Client）
      # 集群模式下，defaultZone应该指向其它Eureka Server，如果有更多其它Server实例，逗号拼接即可
      defaultZone: http://127.0.0.1:8761/eureka,http://127.0.0.1:8762/eureka # 注册到集群汇总，多个用,拼接
    register-with-eureka: true  # 集群模式下可以改成true
    fetch-registry: true # 集群模式下可以改成true



#建议暴露所有的端⼝
management:
  # 暴露健康接口细节
  endpoint:
    health:
      show-details: always

  endpoints:
    web:
      exposure:
        include: "*"

消费者代码和配置都一样，只是服务端口不相同

  <!--eureka client 客户端依赖引⼊-->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>

        <!--spring cloud stream 依赖（rabbit）-->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>

消费配置，以及mq配置

#eureka server服务端口
server:
  port: 7102
spring:
  application:
    name: server-pruduce-stream-server-7102 # 应用名称，应用名称会在Eureka中作为服务名称

  cloud:
    inetutils:
      # 指定此客户端的ip
      default-ip-address: springcloud

    # stream 配置
    stream:
      binders: # 绑定MQ服务信息（此处我们是RabbitMQ）
        testRabbitBinder: # 给Binder定义的名称，⽤于后⾯的关联
          type: rabbit # MQ类型，如果是Kafka的话，此处配置kafka
          environment: # MQ环境配置（⽤户名、密码等）
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 关联整合通道和binder对象
        input: # output是我们定义的通道名称，此处不能乱改
          destination: mqExchange # 要使⽤的Exchange名称（消息队列主题名称）
          content-type: text/plain # application/json # 消息类型设置，⽐如json
          binder: testRabbitBinder # 关联MQ服务
          #group: input001 # 消费者分组 避免多个消费者重复消费同一个任务


eureka:
  instance:
    hostname: springcloud  # 当前eureka实例的主机名
    ip-address: springcloud
  client:
    service-url:
      prefer-ip-address: true
      lease-renewal-interval-in-seconds: 30
      # 租约到期，服务时效时间，默认值90秒,服务超过90秒没有发⽣⼼跳，服务注册中心会将服务从列表移除
      lease-expiration-duration-in-seconds: 90
      # 配置客户端所交互的Eureka Server的地址（Eureka Server集群中每一个Server其实相对于其它Server来说都是Client）
      # 集群模式下，defaultZone应该指向其它Eureka Server，如果有更多其它Server实例，逗号拼接即可
      defaultZone: http://127.0.0.1:8761/eureka,http://127.0.0.1:8762/eureka # 注册到集群汇总，多个用,拼接
    register-with-eureka: true  # 集群模式下可以改成true
    fetch-registry: true # 集群模式下可以改成true



#建议暴露所有的端⼝
management:
  # 暴露健康接口细节
  endpoint:
    health:
      show-details: always

  endpoints:
    web:
      exposure:
        include: "*"

生产者代码

这儿的Source是stream 定义默认的通道，即配置文件中output配置，用来从代码中输出到mq中


public interface Source {
    String OUTPUT = "output";

    @Output("output")
    MessageChannel output();
}

生产者

@EnableBinding(Source.class)
@Component
public class Pruduce {

    // 将MessageChannel的封装对象Source注⼊到这⾥使⽤
    @Autowired
    private Source source;


    public void sendMessage(String content) {
    // 向mq中发送消息（并不是直接操作mq，应该操作的是spring cloud stream）
    // 使⽤通道向外发出消息(指的是Source⾥⾯的output通道)
        source.output().send(MessageBuilder.withPayload(content).build());
    }

}

代码消费

绑定消费通道 input 即是在yml里配置的、bindings 关联整合通道和binder对象，用来配置从哪个mq 队列获取消息，以及消息的类型

默认配置是

public interface Sink {
    String INPUT = "input";

    @Input("input")
    SubscribableChannel input();
}

绑定消费

@EnableBinding(Sink.class)
@Component
public class ConsumerImpl implements Consumer{

    @StreamListener(Sink.INPUT) //用于监听生产者端Message 到来
    @Override
    public void recevieMessages(Message<String> message) {
        System.out.println("=========接收到的消息：" + message);
    }

}

分别启动三个服务

生产者测试

   @Autowired
    private Pruduce pruduce;


    @org.junit.Test
    public void test(){
        pruduce.sendMessage("hello world...");
    }

两个服务消费直接启动服务，使用注解@StreamListener(Sink.INPUT)进行监听某个通道

发现一个生产者两个服务都消费到了

如果不想使用订阅消费者，避免重复消费，可以分组配置，让两个消费者在同一个组里，这样可以避免重复消费。

  group: input001 # 消费者分组 避免多个消费者重复消费同一个任务

由于源码中只有一个通道，在业务中不可能所有的业务使用同一个通道，此时源码中不支持我们的业务场景。

所以，我们需要不同的通道，可以自定义不同的生产者和消费者通道，然后在生产和监听获取的时候，使用我们自己的通道即可。

自定义通道

interface CustomChannel {
String INPUT_LOG = "input1";
String OUTPUT_LOG = "output1";
@Input(INPUT_LOG)
SubscribableChannel inputLog();
@Output(OUTPUT_LOG)
MessageChannel outputLog();
}

生产绑定通道在 @EnableBinding 注解中，绑定⾃定义的接⼝。 @EnableBinding(CustomChannel.class) 然后注入使用 CustomChannel.output().send…

消费监听通道 @StreamListener 做监听的时候，需要指定。

@EnableBinding(CustomChannel.class) @StreamListener(CustomChannel.INPUT_LOG)

转载自:https://juejin.cn/post/7231803423577407546