概述
不同中间件,有各自的使用方法,代码也不一样。
可以使用Spring Cloud Stream解耦,切换中间件时,不需要修改代码。实现方式为使用绑定层,绑定层对生产者和消费者提供统一的编码方式,需要连接不同的中间件时,绑定层使用不同的绑定器即可,也就是把切换中间件需要做相应的修改工作交给绑定层来做。
本文的操作是在 微服务调用链路追踪 的基础上进行。
环境说明
jdk1.8
maven3.6.3
mysql8
spring cloud2021.0.8
spring boot2.7.12
idea2022
rabbitmq3.12.4
步骤
消息生产者
创建子模块stream_producer
添加依赖
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
</dependencies>
刷新依赖
配置application.yml
server:
port: 7001
spring:
application:
name: stream_producer
rabbitmq:
addresses: 127.0.0.1
username: guest
password: guest
cloud:
stream:
bindings:
output:
destination: my-default #指定消息发送的目的地,值为rabbit的exchange的名称
binders:
defaultRabbit:
type: rabbit #配置默认的绑定器为rabbit
查看Source.class
源码
编写生产者代码,发送一条消息("hello world")到rabbitmq的my-default exchange中
package org.example.stream;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
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;
@EnableBinding(Source.class)
@SpringBootApplication
public class StreamProductApplication implements CommandLineRunner {
@Autowired
private MessageChannel output;
@Override
public void run(String... args) throws Exception {
//发送消息
// messageBuilder 工具类,创建消息
output.send(MessageBuilder.withPayload("hello world").build());
}
public static void main(String[] args) {
SpringApplication.run(StreamProductApplication.class, args);
}
}
查看rabbitmq web UI
http://localhost:15672/
看到Exchanges中还没有my-default
运行StreamProductApplication
刷新rabbitmq Web UI,看到了my-dafault的exchange
消息消费者
创建子模块stream_consumer
添加依赖
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
</dependencies>
配置application.yml
server:
port: 7002
spring:
application:
name: stream_consumer
rabbitmq:
addresses: 127.0.0.1
username: guest
password: guest
cloud:
stream:
bindings:
input: #内置获取消息的通道,从destination配置值的exchange中获取信息
destination: my-default #指定消息发送的目的地,值为rabbit的exchange的名称
binders:
defaultRabbit:
type: rabbit #配置默认的绑定器为rabbit
查看内置通道名称为input
编写消息消费者启动类,在启动类监听接收消息
package org.example.stream;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
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;
@SpringBootApplication
@EnableBinding(Sink.class)
public class StreamConsumerApplication {
@StreamListener(Sink.INPUT)
public void input(Message<String> message){
System.out.println("监听收到:" + message.getPayload());
}
public static void main(String[] args) {
SpringApplication.run(StreamConsumerApplication.class, args);
}
}
运行stream_consumer
消费者服务,监听消息
运行stream_producer
生产者服务,发送消息
查看消费者服务控制台日志,接收到了消息
优化代码
之前把生产和消费的消息都写在启动类中了,代码耦合高。
优化思路是把不同功能的代码分开放。
消息生产者
stream_producer 代码结构如下
package org.example.stream.producer;
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 org.springframework.stereotype.Component;
/**
* 向中间件发送数据
*/
@Component
@EnableBinding(Source.class)
public class MessageSender {
@Autowired
private MessageChannel output;//通道
//发送消息
public void send(Object obj){
output.send(MessageBuilder.withPayload(obj).build());
}
}
修改启动类
package org.example.stream;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
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;
@SpringBootApplication
public class StreamProductApplication {
public static void main(String[] args) {
SpringApplication.run(StreamProductApplication.class, args);
}
}
pom.xml添加junit依赖
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
刷新依赖
编写测试类
在stream_producerm模块的src/test目录下,新建org.example.stream
包,再建出ProducerTest类,代码如下
package org.example.stream;
import org.example.stream.producer.MessageSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class ProducerTest {
@Autowired
private MessageSender messageSender;//注入发送消息工具类
@Test
public void testSend(){
messageSender.send("hello world");
}
}
消息消费者
stream_consumer代码结构如下
添加MessageListener类获取消息
package org.example.stream.consumer;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;
@Component
@EnableBinding(Sink.class)
public class MessageListener {
// 监听binding中的信息
@StreamListener(Sink.INPUT)
public void input(String message){
System.out.println("获取信息:" + message);
}
}
修改启动类
package org.example.stream;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
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;
@SpringBootApplication
public class StreamConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(StreamConsumerApplication.class, args);
}
}
启动consumer接收消息
执行producer单元测试类ProducerTest的testSend()方法,发送消息
查看consumer控制台输出,接收到信息了
代码解耦后,同样能成功生产消息和消费消息。
自定义消息通道
此前使用默认的消息通道output
和input。
也可以自己定义消息通道,例如:myoutput
和myinput
消息生产者
在org.example.stream
包下新建channel
包,在channel
包下新建MyProcessor接口类
package org.example.stream.channel;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* 自定义的消息通道
*/
public interface MyProcessor {
/**
* 消息生产这的配置
*/
String MYOUTPUT = "myoutput";
@Output("myoutput")
MessageChannel myoutput();
/**
* 消息消费者的配置
*/
String MYINPUT = "myinput";
@Input("myinput")
SubscribableChannel myinput();
}
修改MessageSender
类
package org.example.stream.producer;
import org.example.stream.channel.MyProcessor;
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 org.springframework.stereotype.Component;
/**
* 向中间件发送数据
*/
@Component
@EnableBinding(MyProcessor.class)
public class MessageSender {
@Autowired
private MessageChannel myoutput;//通道
//发送消息
public void send(Object obj){
myoutput.send(MessageBuilder.withPayload(obj).build());
}
}
修改application.yml
cloud:
stream:
bindings:
output:
destination: my-default #指定消息发送的目的地
myoutput:
destination: custom-output
消息消费者
在stream_consumer服务的org.example.stream
包下新建channel
包,在channel
包下新建MyProcessor接口类
package org.example.stream.channel;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* 自定义的消息通道
*/
public interface MyProcessor {
/**
* 消息生产者的配置
*/
String MYOUTPUT = "myoutput";
@Output("myoutput")
MessageChannel myoutput();
/**
* 消息消费者的配置
*/
String MYINPUT = "myinput";
@Input("myinput")
SubscribableChannel myinput();
}
修改MessageListener
类
package org.example.stream.stream;
import org.example.stream.channel.MyProcessor;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;
@Component
@EnableBinding(MyProcessor.class)
public class MessageListener {
// 监听binding中的信息
@StreamListener(MyProcessor.MYINPUT)
public void input(String message){
System.out.println("获取信息:" + message);
}
}
修改application.yml配置
cloud:
stream:
bindings:
input: #内置获取消息的通道,从destination配置值的exchange中获取信息
destination: my-default #指定消息发送的目的地
myinput:
destination: custom-output
测试
启动stream_consumer
运行单元测试的testSend()方法生产消息
查看stream_consumer
控制台,能看到生产的消息,如下
获取信息:hello world
消息分组
采用复制配置方式运行两个消费者
启动第一个消费者(端口为7002)
修改端口为7003,copy configuration,再启动另一个消费者
执行生产者单元测试生产消息,看到两个消费者都接收到了信息
说明:如果有两个消费者,生产一条消息后,两个消费者均能收到信息。
但当我们发送一条消息只需要其中一个消费者消费消息时,这时候就需要用到消息分组,发送一条消息消费者组内只有一个消费者消费到。
我们只需要在服务消费者端设置spring.cloud.stream.bindings.input.group
属性即可
重启两个消费者
修改端口号为7002,重新启动第一个消费者
修改端口号为7003,重新启动第二个消费者
生产者生产一条消息
查看消费者接收消息情况,只有一个消费者接收到信息。
消息分区
消息分区就是实现特定消息只往特定机器发送。
修改生产者配置
cloud:
stream:
bindings:
output:
destination: my-default #指定消息发送的目的地,值为rabbit的exchange的名称
myoutput:
destination: custom-output
producer:
partition-key-expression: payload #分区关键字 可以是对象中的id,或对象
partition-count: 2 #分区数量
修改消费者1的application.yml配置
server:
port: 7002
spring:
application:
name: stream_consumer
rabbitmq:
addresses: 127.0.0.1
username: guest
password: guest
cloud:
stream:
bindings:
input: #内置获取消息的通道,从destination配置值的exchange中获取信息
destination: my-default #指定消息发送的目的地,值为rabbit的exchange的名称
myinput:
destination: custom-output
group: group1 #消息分组,有多个消费者时,只有一个消费者接收到信息
consumer:
partitioned: true #开启分区支持
binders:
defaultRabbit:
type: rabbit #配置默认的绑定器为rabbit
instance-count: 2 #消费者总数
instance-index: 0 #当前消费者的索引
启动消费者1
修改消费者2的配置
server:
port: 7003
spring:
application:
name: stream_consumer
rabbitmq:
addresses: 127.0.0.1
username: guest
password: guest
cloud:
stream:
bindings:
input: #内置获取消息的通道,从destination配置值的exchange中获取信息
destination: my-default #指定消息发送的目的地,值为rabbit的exchange的名称
myinput:
destination: custom-output
group: group1 #消息分组,有多个消费者时,只有一个消费者接收到信息
consumer:
partitioned: true #开启分区支持
binders:
defaultRabbit:
type: rabbit #配置默认的绑定器为rabbit
instance-count: 2 #消费者总数
instance-index: 1 #当前消费者的索引
修改端口号为7003,当前消费者的索引instance-index
的值修改为1
启动消费者2
生产者发送消息,看到只有Application(2)接收到消息
再用生产者发送一次消息,也是Application(2)接收到消息
说明实现了消息分区
也可以更改发送的数据,看是否能发送到不同消费者
修改生产者,发送数据由hello world
变为hello world1
,同时发送5次
public void testSend(){
for (int i = 0; i < 5; i++) {
messageSender.send("hello world1");
}
}
看到hello world1
全部被Application消费
所以消息分区是根据发送的消息不同,发送到不同消费者中。
完成!enjoy it!