SpringCloud - Spring Cloud 之 Stream构建消息驱动微服务框架（十九）

Posted 2022-07-11 MinggeQingchun

阅读本文前可先参考

​​​​​​SpringCloud - Spring Cloud根/父项目，开发准备（二）_MinggeQingchun的博客-CSDN博客

在微服务的开发过程中，经常用到消息中间件，通过消息中间件在服务与服务之间传递消息，不管使用哪款消息中间件，如RabbitMQ还是Kafka，那么消息中间件和服务之间都有耦合性，如原来使用的RabbitMQ，现在要替换为Kafka，那么我们的微服务都需要修改，变动会比较大。或者 服务A用的是RabbitMQ，服务B用的是Kafka，我们能否不关注MQ底层实现，进行适配绑定

Spring Cloud Stream就是负责整合我们的消息中间件，降低微服务和消息中间件的耦合性，做到轻松在不同消息中间件间切换

注：

目前Spring Cloud Stream仅支持RabbitMQ、Kafka

Spring Cloud Stream

Spring Cloud Stream 是一个构建消息驱动微服务的框架

应用程序通过input（相当于消费者consumer）、output（相当于生产者producer）来与Spring Cloud Stream中Binder交互，而Binder负责与消息中间件交互，因此，我们只需关注如何与Binder交互即可，而无需关注与具体消息中间件的交互。 

组成	说明
Binder	Binder是应用与消息中间件之间的封装，目前实现了Kafka和RabbitMQ的Binder，通过Binder可以很方便的连接中间件，可以动态的改变消息类型(对应于Kafka的topic，RabbitMQ的exchange)，这些都可以通过配置文件来实现；
@Input	该注解标识输入通道，通过该输入通道接收消息进入应用程序
@Output	该注解标识输出通道，发布的消息将通过该通道离开应用程序
@StreamListener	监听队列，用于消费者的队列的消息接收
@EnableBinding	将信道channel和exchange绑定在一起

1、Spring Cloud Stream应用

1、消息生产者

1、新建一个springboot Module（springcloud-10-service-stream-producer），设置父项目

2、添加 spring-cloud-starter-stream-rabbit等 依赖

<dependencies>
        <!-- spring-cloud-starter-stream-rabbit -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
    </dependencies>

3、 application.prperties配置文件

#对接具体的消息中间件（rabbitmq是一个自定义的key，map集合）
spring.cloud.stream.binders.rabbitmq.type=rabbit
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.host=192.168.133.129
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.username=root
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.password=root
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.virtual-host=/

#消息生产者
#其中output是一个key，这个名字是一个通道的名称，在代码中会用到
#destination表示要使用的Exchange名称定义（output是一个自定义的key，map集合；此处使用rabbitmq，因此设置的是交换机）
spring.cloud.stream.bindings.output.destination=spring.cloud.stream.exchange
#设置要绑定的消息服务的binder
spring.cloud.stream.bindings.output.binder=rabbitmq

点击 binders，进入 BindingServiceProperties 类

点击 type，environment ，进入 BinderProperties 类

点击 bindings，进入 BindingServiceProperties 类

 

4、消息发送类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;

import java.util.Date;

/**
 * 消息发送类
 */
//此处MQ使用的RabbitMQ，因此是将信道channel和exchange绑定在一起
//Source：提供消息发布的目的地，是消息的生产者
@EnableBinding(Source.class)
public class MessageSender 

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

    public void publish(String msg)
        output.send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送----<消息内容：" + msg + ">，发送时间：" + new Date());
    




@Service
public class MessageService 
    @Autowired
    private MessageSender messageSender;

    public void sendMessage(String msg) 
        messageSender.publish(msg);
    




@SpringBootApplication
public class Stream10ProducerApplication 
    public static void main(String[] args) 
        ApplicationContext context = SpringApplication.run(Stream10ProducerApplication.class, args);
        MessageService messageService = context.getBean(MessageService.class);
        messageService.sendMessage("Hello Spring Cloud Stream!");
    

2、消息消费者

1、新建一个springboot Module（springcloud-10-service-stream-consumer），设置父项目

2、添加 spring-cloud-starter-stream-rabbit等 依赖

<dependencies>
        <!-- spring-cloud-starter-stream-rabbit -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
    </dependencies>

3、application.prperties配置文件

#对接具体的消息中间件（rabbitmq是一个自定义的key，map集合）
spring.cloud.stream.binders.rabbitmq.type=rabbit
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.host=192.168.133.129
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.port=5672
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.username=root
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.password=root
spring.cloud.stream.binders.rabbitmq.environment.spring.rabbitmq.virtual-host=/


#配置消息消费者
#指定交换机
spring.cloud.stream.bindings.input.destination=spring.cloud.stream.exchange
#设置要绑定的消息服务的binder
spring.cloud.stream.bindings.input.binder=rabbitmq

4、消息接收类

//Sink：提供消息消费的目的地，是消息的消费者
@EnableBinding(Sink.class)
public class MessageReceiver 

    //监听队列，用于消费者的队列的消息接收
    @StreamListener(Sink.INPUT)
    public void receiveMessage(Message msg)
        System.out.println("消息接收----<消息内容：" + msg.getPayload() + ">，接收时间：" + new Date());
    

2、Spring Cloud Stream自定义消息通道

消息传递主要使用的是系统提供的 Source （output）、Sink（input）；因此我们自定义Source和Sink接口即可

1、消息生产者

1、自定义Source接口

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

/**
 * 仿写 Source 接口
 */
public interface MessageSource 
    //自定义 channel名称（这里"myOutput" 要和 application.properties 配置文件中的 spring.cloud.stream.bindings.myOutput 的key 保持一致）
    String OUTPUT = "myOutput";
    //系统默认 channel名称
//    String OUTPUT = "output";

    //@Output注解标识输出通道，发布的消息将通过该通道离开应用程序
    @Output(MessageSource.OUTPUT)
//    @Output("output")
    MessageChannel output();




import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.support.MessageBuilder;

import java.util.Date;

/**
 * 消息发送类
 */
//此处MQ使用的RabbitMQ，因此是将信道channel和exchange绑定在一起
//Source：提供消息发布的目的地，是消息的生产者
@EnableBinding(MessageSource.class)
public class MessageCustomSender 

    //消息的发送管道
    @Autowired
    private MessageSource messageSource;

    public void publish(String msg)
        messageSource.output().send(MessageBuilder.withPayload(msg).build());
        System.out.println("消息发送----<消息内容：" + msg + ">，发送时间：" + new Date());
    

2、application.properties配置文件修改

#消息生产者
#其中output是一个key，这个名字是一个通道的名称，在代码中会用到
#destination表示要使用的Exchange名称定义（output是一个自定义的key，map集合；此处使用rabbitmq，因此设置的是交换机）
#（自定义名称 myOutput，这里的 myOutput 要和 Source 接口中的 String OUTPUT = "myOutput"; 保持一致）
spring.cloud.stream.bindings.myOutput.destination=spring.cloud.stream.exchange
#spring.cloud.stream.bindings.output.destination=spring.cloud.stream.exchange
#设置要绑定的消息服务的binder（自定义名称 myOutput）
spring.cloud.stream.bindings.myOutput.binder=rabbitmq
#spring.cloud.stream.bindings.output.binder=rabbitmq

注：application.properties配置文件中的 spring.cloud.stream.bindings.myOutput 要和 Source 接口中的 String OUTPUT = "myOutput"; 保持一致

2、消息消费者

1、自定义Sink接口类

/**
 * 仿写 Sink 接口
 */
public interface MessageSink 
    //自定义 channel名称（这里"myInput" 要和 application.properties 配置文件中的 spring.cloud.stream.bindings.myInput 的key 保持一致）
    String INPUT = "myInput";
    //系统默认 channel名称
//    String INPUT = "input";

    //@Input注解标识输入通道，通过该输入通道接收消息进入应用程序
    @Input(MessageSink.INPUT)
//    @Input("input")
    SubscribableChannel input();




//Sink：提供消息消费的目的地，是消息的消费者
@EnableBinding(MessageSink.class)
public class MessageCustomReceiver 

    //监听队列，用于消费者的队列的消息接收
    @StreamListener(MessageSink.INPUT)
    public void receiveMessage(Message msg)
        System.out.println("消息接收----<消息内容：" + msg.getPayload() + ">，接收时间：" + new Date());
    

2、application.properties配置文件修改

#配置消息消费者
#指定交换机（自定义名称 myInput，这里的 myInput 要和 Sink 接口中的 String INPUT = "myInput"; 保持一致）
spring.cloud.stream.bindings.myInput.destination=spring.cloud.stream.exchange
#spring.cloud.stream.bindings.input.destination=spring.cloud.stream.exchange
#设置要绑定的消息服务的binder（自定义名称 myInput）
spring.cloud.stream.bindings.myInput.binder=rabbitmq
#spring.cloud.stream.bindings.input.binder=rabbitmq

注：application.properties配置文件中的 spring.cloud.stream.bindings.myInput 要和 Sink 接口中的 String INPUT = "myInput"; 保持一致

3、Spring Cloud Stream分组与持久化

上述两种消息的发送与接收，在默认情况下都属于一种临时消息，如果没有消费者进行消费处理，那么该消息是不会被永久保留，可能会造成消息的丢失

消息消费者

如果要实现持久化消息，需要在消息的消费端配置一个分组，有分组就表示该消息可以进行持久化

在application.properties配置文件 （仅在消费端设置）

#指定分组，可以进行消息的持久化 applies to consumers only
spring.cloud.stream.bindings.myInput.group=rabbitmq-group

在Spring Cloud Stream 中在消费者端如果将队列设置为持久化队列，则队列名称会变为为destination.group，此时消费端的微服务宕机或重启，该队列信息依然会被保留在 RabbitMQ中，后续依然可以进行消费 

消息分组：（还可以实现同一分组只有一个消费者能接收到消费）

没有做分组时，一个消息可以被多个消费者接收，分组可以让一个消息只能被一个消费者接收，避免一个消息被多个消费者消费；

当项目集群部署了很多份，那么就会变成多个消费者，但是业务可能需要的是一个消息只消费一次，所以此时需要加个分组，就可以实现同一个分组里面的消费者只会有一个消费者能接收到消息；

注：

1、不分组的话，消费者要先启动起来，然后再用生产者发送消息，这样才可以接收到消息，否则发送的消息就丢失了，生产者先发了消息，消费者后面才启动的话是接收不到消息的；

2、不分组的话，多个消费者都能接收消息，也就是一个消息可以被多个消费者接收；

4、Spring Cloud Stream设置路由键

默认情况下Spring Cloud Stream传送消息属于广播消息，默认匹配方式是 #，表示所有消费者都可以匹配上，我们也可以通过指定路由键 RoutingKey实现按需求匹配消息消费端进行消息接收处理

消息消费者

在 application.properties配置文件中

#设置一个RoutingKey路由key，默认是#，我们可以指定
spring.cloud.stream.rabbit.bindings.myInput.consumer.bindingRoutingKey=spring.cloud.stream.#

由此开发使用RabbitMQ有两种选择

1、SpringBoot + RabbitMQ 整合实现消息传送；

2、使用Spring Cloud Stream对消息中间件的包装，来实现消息传送

高性能云服务器 精品线路独享带宽，毫秒延迟，年中盛惠 1 折起