当前位置：首页 > news >正文

消息驱动Stream---基于SpringCloud

news 来源：原创 2024/9/21 23:28:25

概要：实际开发中，服务与服务之间的通信经常会使用到消息中间件，而以往使用的一些消息中间件，比如RabbitMQ，该中间件和系统的耦合性非常高，如果我们要将RabbitMQ替换为Kafka，那么系统将会有较大的变动。此时，我们可以使用Spring Cloud Stream整合消息中间件降低系统和中间件的耦合性。

Spring Cloud Stream简介

什么是Spring Cloud Stream

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

作用：它是Spring Cloud对于消息中间件的进一步封装，通过使用Spring Cloud Stream，可忽略消息中间件之间的差异，有效降低开发人员对消息中间件的使用复杂度。目前Spring Cloud Stream支持的消息中间件仅有RabbitMQ和Kafka。

Spring Cloud Stream 与消息中间件的交互

使用Stream构建的应用程序与消息中间件之间是通过绑定器Binder相关联的。

Binder

特点：Binder对于应用程序而言起到了隔离作用，它使得不同消息中间件的实现细节对Stream应用程序来说是透明的。

用途：对于每一个Stream的应用程序来说，Binder无需知晓消息中间件的通信细节，而是通过向应用程序暴露统一的通道(Channel)来进行通信。Binder是作为输入通道（inputs）和输出通道（outputs）与消息中间件之间的桥梁进行消息通信。

构建Stream工程

快速构建Stream工程

1.创建stream-hello项目

加入依赖：

    <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId><version>2.1.3.RELEASE</version></dependency></dependencies>

创建rabittmq包，并在rabittmq包中创建SinkReceiver类，用于接收RabbitMQ发送的消息。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;@EnableBinding(Sink.class)  //开启绑定通道的注解  Sink是Stream组件默认的输入通道接口
public class SinkReceiver {//声明日志private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class);@StreamListener(Sink.INPUT)      //此注解声明此方法为监听方法private void receiver(String payload){logger.info("Receiver:"+payload);}
}

2.编写消息消费者类

@EnableBinding注解：用来指定一个或多个定义了@Input或@Output注解的接口，以此实现对消息通道(Channel）的绑定。

@StreamListener注解：主要是修饰方法，用于将被修饰的方法注册为消息中间件上数据流的事件监听器，注解中的属性值对应了监听的消息通道名。

3.项目测试

输入Rabbit MQ的可视化网址，账号密码默认为：guest

第一个Spring Cloud Stream案例就完成了，但是我们会发现，此案例中并没有在配置文件 application.yml中进行任何属性设置，原因在于Spring Cloud Stream会为消息中间件RabbitMQ提供默认的自动化配置。当然我们也可以在Spring Boot支持的全局配置文件application.properties或application.yml中修改相关配置。

Stream的发布-订阅模式

Stream的分布-订阅

特点：Spring Cloud Stream中的消息通信方式遵循的是发布-订阅模式，当一条消息被投递到消息中间件之后，它会通过共享主题的方式进行广播，消息消费者在订阅的主题中收到它并触发自身的业务逻辑处理。这里所提到的主题是Spring Cloud Stream中的一个抽象概念，用来代表发布共享消息给消费者的地方

Stream框架应用结构图

提供者发送消息到RabbitMQ等消息中间件，消费者通过订阅的方式从消息中间件获取消息。

搭建工程实现Stream的发布-订阅

1.启动rabbitmq

2.创建提供者

依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId><version>2.1.3.RELEASE</version></dependency>

application.yml文件：

server:port: 8898
spring:application:name: stream-rabbitmq-providerrabbitmq:host: localhostport: 5672username: guestpassword: guestcloud:stream:bindings:output:destination: minestream

创建StreamProvider


import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@EnableBinding(Source.class)
@RestController
public class StreamProvider {@Autowired@Output(Source.OUTPUT)private MessageChannel channel;@GetMapping("/send")public void send(){channel.send(MessageBuilder.withPayload("hello world").build());}
}

3.创建消费者

依赖和上面提供者一样（复制即可）

application.yml:

server:port: 9898
spring:application:name: stream-rabbitmq-consumerrabbitmq:host: localhostport: 5672username: guestpassword: guestcloud:stream:bindings:input:destination: minestream

创建StreamConsumer

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;@EnableBinding(Sink.class)
public class StreamConsumer {@StreamListener(Sink.INPUT)public void receiver(String payload){System.out.println("接收到了mq中发送过来的消息"+payload);}
}

4.测试运行

访问http://localhost:8898/send 地址发送消息，在消费者项目stream-rabbitmq-consumer的控制台可以看到打印日志“接收到MQ消息：Hello World!”，说明消息已成功被接收。

Stream的消费组和消息分区

Stream消费组的实现

在Spring Cloud Stream应用程序开发中，如果在同一主题上的应用需要启动多个实例时，为防止对消息的重复处理，我们可以通过spring.cloud.stream.bindings.input. group属性为应用指定一个组名，这样一个应用的多个实例在接收到消息时，只会有一个实例真正收到消息并进行处理。