【Hoxton.SR1版本】SpringCloudStream消息驱动

目录

一、简介

二、搭建消息生产者端

三、搭建消息消费者端

四、消息重复消费问题

五、消息持久化

六、总结

一、简介

在实际项目中&＃xff0c;服务与服务之间的通信往往我们会采用消息中间件方式来处理&＃xff0c;比如引入RabbitMQ、Kafka等&＃xff0c;但这会有一个问题&＃xff0c;就是我们的应用程序跟消息中间件耦合在一块了&＃xff0c;还有就是如果我们要替换为Kafka&＃xff0c;那么变动会比较大&＃xff0c;Spring Cloud官网提供了Spring Cloud Stream组件&＃xff0c;用来给我们整合消息中间件&＃xff0c;Spring Cloud Stream底层屏蔽了消息中间件的差异&＃xff0c;降低了切换成本&＃xff0c;统一消息的编程模型&＃xff0c;这样就可以降低我们系统和消息中间件的耦合度。总结一句话&＃xff1a;就是Spring Cloud Stream有利于应用程序与消息中间件的解耦。

• Spring Cloud Stream是什么&＃xff1f;

官方定义Spring Cloud Stream是一个构建消息驱动微服务的框架。应用程序通过inputs和outputs来与Spring Cloud Stream中binder对象交互。通过我们配置来binding(绑定)&＃xff0c;而Spring Cloud Stream的binder对象负责与消息中间件交互。所以&＃xff0c;我们只需要搞清楚如何与Spring Cloud Stream交互就可以方便使用消息驱动的方式。

通过Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream为一些供应商的消息中间件产品提供了个性化的自动化配置实现&＃xff0c;引用了发布-订阅&＃xff0c;消费组、分区的三个核心概念。但是目前 Spring Cloud Stream 只支持 RabbitMQ 和 Kafka 的自动化配置。

• Spring Cloud Stream官方文档地址

https://spring.io/projects/spring-cloud-stream

https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.6.RELEASE/reference/html/

• Spring Cloud Stream官网结构图

可以看到&＃xff0c;通过定义绑定器Binder作为中间层&＃xff0c;实现了应用程序与消息中间件细节之间的隔离。

• Spring Cloud Stream编码API和常用注解

1. Middleware&＃xff1a;中间件&＃xff0c;目前只支持RabbitMQ和Kafka。
2. Binder&＃xff1a;Binder是应用与消息中间件之间的封装&＃xff0c;目前实行了Kafka和RabbitMQ的Binder&＃xff0c;通过Binder可以很方便地连接中间件&＃xff0c;可以动态的改变消息类型(对应于Kafka的topic&＃xff0c;RabbitMQ的exchange)&＃xff0c;这些都可以通过配置文件来实现。
3. &＃64;Input&＃xff1a;注解标识输入通道&＃xff0c;通过该输入通道接收到的消息进入应用程序。
4. &＃64;Output&＃xff1a;注解标识输出通道&＃xff0c;发布的消息将通过该通道离开应用程序。
5. &＃64;StreamListener&＃xff1a;监听队列&＃xff0c;用于消费者的队列的消息接收。
6. &＃64;EnableBinding&＃xff1a;指信道channel和exchange绑定在一起。

• Stream几个重要概念

1. Destination Binders&＃xff1a;目标绑定器&＃xff0c;目标指的是 kafka 还是 RabbitMQ&＃xff0c;绑定器就是封装了目标中间件的包。如果操作的是 kafka 就使用 kafka binder &＃xff0c;如果操作的是 RabbitMQ 就使用 rabbitmq binder&＃xff1b;
2. Destination Bindings&＃xff1a;外部消息传递系统和应用程序之间的桥梁&＃xff0c;提供消息的“生产者”和“消费者”&＃xff08;由目标绑定器创建&＃xff09;&＃xff1b;
3. Message&＃xff1a;一种规范化的数据结构&＃xff0c;生产者和消费者基于这个数据结构通过外部消息系统与目标绑定器和其他应用程序通信&＃xff1b;

二、搭建消息生产者端

新建一个module【springcloud-stream-rabbitmq-provider8801】

【a】pom.xml&＃xff1a;引入spring-cloud-starter-stream-rabbit依赖


springcloud2020com.wsh.springcloud1.0-SNAPSHOT4.0.0springcloud-stream-rabbitmq-provider8801org.springframework.cloudspring-cloud-starter-stream-rabbitorg.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-actuatororg.springframework.cloudspring-cloud-starter-netflix-eureka-clientorg.springframework.bootspring-boot-starter-testtest


【b】application.yml&＃xff1a;相关配置都在配置文件里面做了较详细的说明

server:port: 8801
spring:application:name: springcloud-stream-rabbitmq-providercloud:stream:binders: # 在此处配置要绑定的rabbitmq的服务信息rabbitmq_binder: # binder绑定器名称&＃xff0c;用于binding整合type: rabbit # 消息组件类型 如果消息中间件是kafka,则type:kafkaenvironment: # rabbitmq相关环境配置spring:rabbitmq:host: localhost #rabbitmq主机port: 5672 #rabbitmq端口username: guest #rabbitmq用户名password: guest #rabbitmq用户密码bindings: # 服务的整合处理output: # 输出通道,表示消息生产方destination: rabbitmq_stream_exchange # 指定输出的交换器名称content-type: application/json # 设置消息类型&＃xff0c;本次为json&＃xff0c;文本则设置“text/plain”binder: rabbitmq_binder # 指定binder的名称,需与上面spring.cloud.stream.binders.xxx中的xxx绑定器名称对应
eureka:client:service-url:defaultZone: http://springcloud-eureka7001.com:7001/eureka/,http://springcloud-eureka7002.com:7002/eureka/ #集群版Eureka注册中心

【c】主启动类

&＃64;SpringBootApplication
public class SpringCloudStreamMQServiceApplicaiton8801 {public static void main(String[] args) {SpringApplication.run(SpringCloudStreamMQServiceApplicaiton8801.class, args);}
}

【d】定义消息发送的接口

/*** &＃64;Description 消息发送接口* &＃64;Date 2020/8/27 21:37* &＃64;Author weishihuai* 说明:*/
public interface IMessageProvider {/*** 发送消息*/String sendMessage();
}

消息发送实现类&＃xff1a;

package com.wsh.springcloud.service.impl;import com.wsh.springcloud.service.IMessageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;/*** &＃64;Description 消息发送实现类* &＃64;Date 2020/8/27 21:38* &＃64;Author weishihuai* 说明: &＃64;EnableBinding表示信道channel和exchange绑定在一起.*/
&＃64;EnableBinding(Source.class)
public class MessageProviderImpl implements IMessageProvider {private static final Logger logger &＃61; LoggerFactory.getLogger(MessageProviderImpl.class);/*** 消息发送管道*/&＃64;Resourceprivate MessageChannel output;&＃64;Overridepublic String sendMessage() {String uuid &＃61; UUID.randomUUID().toString();output.send(MessageBuilder.withPayload(uuid).build());logger.info("消息发送者发送消息: {}", uuid);return "消息发送者发送消息: " &＃43; uuid;}}


 【e】定义消息发送Controller

package com.wsh.springcloud.controller;import com.wsh.springcloud.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;/*** &＃64;Description 消息发送测试Controller* &＃64;Date 2020/8/27 21:39* &＃64;Author weishihuai* 说明:*/
&＃64;RestController
public class SendMessageController {&＃64;Resourceprivate IMessageProvider messageProvider;&＃64;GetMapping(value &＃61; "/sendMessage")public String sendMessage() {return messageProvider.sendMessage();}}


【f】测试

启动Eureka注册中心以及消息驱动发送方服务&＃xff0c;浏览器访问&＃xff1a;http://localhost:8801/sendMessage 测试发送消息&＃xff0c;然后我们去RabbitMQ界面观察流量情况&＃xff1a;

注意&＃xff1a;下图中的rabbitmq_stream_exchange就是我们在application.yml中指定的将消息输出到哪个desitination交换机上面。

观察后台日志&＃xff1a;

 可见&＃xff0c;消息成功发送到MQ中&＃xff0c;正在等待消费方进行消费消息&＃xff0c;至此&＃xff0c;消息发送者端搭建成功&＃xff0c;接下来搭建消息消费方服务。

三、搭建消息消费者端

新建module【springcloud-stream-rabbitmq-consumer8802】

【a】pom.xml


springcloud2020com.wsh.springcloud1.0-SNAPSHOT4.0.0springcloud-stream-rabbitmq-consumer8802org.springframework.cloudspring-cloud-starter-stream-rabbitorg.springframework.bootspring-boot-starter-weborg.springframework.cloudspring-cloud-starter-netflix-eureka-clientorg.springframework.bootspring-boot-starter-actuatororg.springframework.bootspring-boot-starter-testtest


【b】applicaiton.yml

server:port: 8802
spring:application:name: springcloud-stream-rabbitmq-consumercloud:stream:binders:rabbitmq_binder: # binder绑定器名称&＃xff0c;用于binding整合type: rabbit # 消息组件类型 如果消息中间件是kafka,则type:kafkaenvironment: # 设置rabbitmq的相关的环境配置spring:rabbitmq:host: localhost #rabbitmq主机port: 5672 #rabbitmq端口username: guest #rabbitmq用户名password: guest #rabbitmq用户密码bindings: # 服务的整合处理input: # 输入通道,表示消息消费方destination: rabbitmq_stream_exchange # 指定接收的交换器名称,需与消息发送方的destination对应上content-type: application/json # 设置消息类型&＃xff0c;本次为对象json&＃xff0c;如果是文本则设置“text/plain”binder: rabbitmq_binder # 指定binder的名称,需与上面spring.cloud.stream.binders.xxx对应中的xxx对应
eureka:client:service-url:defaultZone: http://springcloud-eureka7001.com:7001/eureka/,http://springcloud-eureka7002.com:7002/eureka/ #集群版Eureka注册中心

【c】主启动类

&＃64;SpringBootApplication
public class RabbitMQStreamServiceApplication8802 {public static void main(String[] args) {SpringApplication.run(RabbitMQStreamServiceApplication8802.class, args);}
}

【d】新增接收消息发送方发送消息的方法

package com.wsh.springcloud.controller;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;/*** &＃64;version V1.0* &＃64;ClassName: com.wsh.springcloud.controller.ReceiveMessageController.java* &＃64;Description: 接收消息发送方发送的消息* &＃64;author: weishihuai* &＃64;date: 2020/8/28 10:55*/
&＃64;Component
&＃64;EnableBinding(Sink.class)
public class ReceiveMessageController {private static final Logger logger &＃61; LoggerFactory.getLogger(ReceiveMessageController.class);&＃64;Value("${server.port}")private String serverPort;/*** 接收消息发送方发送的消息** &＃64;param message 消息* &＃64;StreamListener 通过&＃64;StreamListener注解来监听exchange中的消息*/&＃64;StreamListener(Sink.INPUT)private void receiveMessage(Message message) {String payload &＃61; message.getPayload();logger.info("消息接收方接收消息: {}, 服务端口号&＃xff1a;{}", payload, serverPort);}}


 【e】测试

启动消息消费端&＃xff0c;浏览器访问两次&＃xff1a;http://localhost:8801/sendMessage 模拟发送两条消息到RabbitMQ中&＃xff0c;查看消费者端是否成功消费此消息。

下图是消息发送方的日志&＃xff1a;

下图是消息消费方的日志&＃xff1a;

由此可见&＃xff0c;成功实现了消息接收者获取到了发送者发送的消息&＃xff0c;同时我们在RabbitMQ的web界面也可以看到相关的信息&＃xff1a;

四、消息重复消费问题

为了模拟消息重复消费的问题&＃xff0c;这里我们还需要一个消息消费端&＃xff0c;所以我们新建一个module【springcloud-stream-rabbitmq-consumer8803】&＃xff0c;此子模块跟【springcloud-stream-rabbitmq-consumer8802】除了端口号&＃xff0c;其他一模一样&＃xff0c;这里不再过多阐述。

启动8803消费者和8802消费者&＃xff0c;浏览器访问两次&＃xff1a;http://localhost:8801/sendMessage&＃xff0c;模拟发送两条消息。

(1)、消息发送端日志

(2)、消息接收端【8802】日志

(3)、消息接收端【8803】日志

由此可见&＃xff0c;同一条消息同时被两个消费者处理&＃xff0c;这是不对的。

比如在如下场景中&＃xff0c;假如订单服务调用支付服务&＃xff0c;支付服务我们做集群部署&＃xff0c;那如果支付服务重复消费了订单服务发送过来的支付消息&＃xff0c;那么就会造成数据错误&＃xff0c;我们得避免这种情况。试想一下重复扣用户的款&＃xff0c;这肯定不行的。

接下来&＃xff0c;我们谈谈怎么利用Stream来处理重复消费的问题。Spring Cloud Stream提供了Group组的概念&＃xff0c;我们可以使用Stream中的消息分组来解决。

注意在Stream中处于同一个group中的多个消费者是竞争关系&＃xff0c;就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)。

导致原因&＃xff1a;默认分组group是不同的&＃xff0c;组流水号不一样&＃xff0c;被认为不同组&＃xff0c;可以消费。

从RabbitMQ可视化界面中&＃xff0c;我们可以看到【8802】和【8803】被分配的默认分组信息&＃xff1a;

可以看到&＃xff0c;两个消费者的group组名是不一样的&＃xff0c;所以导致了重复消费。Spring Cloud Stream提供了自定义分组配置的功能&＃xff0c;我们可以将【8802】和【8803】分配相同的组名&＃xff0c;具体配置如下&＃xff1a;

 在【8802】和【8803】的application.yml配置文件中都加入&＃xff1a; group&＃xff1a;group1  指定相同的分组名称&＃xff0c;如下图所示&＃xff1a;

重启【8802】和【8803】&＃xff0c;浏览器访问两次&＃xff1a;http://localhost:8801/sendMessage  模拟发送两条消息

(1)、消息发送端日志

(2)、消息接收端【8802】日志

(3)、消息接收端【8803】日志

可以看到&＃xff0c;同一条消息同时只能被一个消费者处理&＃xff0c;成功防止了消息的重复消费问题。同时我们在RabbitMQ的web界面也可以看到相关的信息&＃xff1a;

五、消息持久化

除了使用group能防止消息重复消费&＃xff0c;其实group还能将消息进行持久化&＃xff0c;下面我们来测试一下。

(1)、停掉【8802】和【8803】两个消息消费者服务

(2)、注释掉【8802】服务中的group分组属性&＃xff0c;注意【8803】需要保留group分组属性

(3)、浏览器访问两次http://localhost:8801/sendMessage&＃xff0c;模拟发送两条消息。

     (a)、消息发送端日志

接着我们重启【8802】和【8803】服务&＃xff0c;注意观察后台日志&＃xff1a;

     (b)、消息接收方【8802】日志

可见&＃xff0c;没有消息消费的日志信息。 

     (c)、消息接收方【8803】日志 

可以看到&＃xff0c;保留group属性的【8803】服务实现了对消息的持久化&＃xff0c;当重启之后会自动去拉取未消费的消息来进行消费&＃xff1b;而【8802】由于未保留group属性&＃xff0c;所以并没有重新去拉取最新消息进行消费。 

六、总结

本篇文章总结了如何使用Spring Cloud Stream消息驱动屏蔽消息中渐渐的底层实现&＃xff0c;极大地方便我们开发者。同时讲解了如何使用分组来避免消息重复消费的问题以及消息持久化。Spring Cloud Stream实现了消息中间件和应用程序的高度解耦。以上相关项目的代码我已经放在Gitee上&＃xff0c;有需要的小伙伴可以去拉取进行学习&＃xff1a;https://gitee.com/weixiaohuai/springcloud_Hoxton&＃xff0c;由于笔者水平有限&＃xff0c;如有不对之处&＃xff0c;还请小伙伴们指正&＃xff0c;相互学习&＃xff0c;一起进步。