我们知道，当微服务越来越来多的时候，仅仅是feign的http调用方式已经满足不了我们的使用场景了。这个时候系统就需要接入消息中间件了。相比较于传统的Spring项目、SpringBoot项目使用消息中间件的很多配置不同，SpringCloud Stream抽象了中间件产品的不同，在SpringCloud中你仅仅需要修改几行配置文件就可以灵活的切换中间件产品而不需要修改任何代码。
现在我们以SpringCloud Stream整合RabbitMQ为例来学习一下
创建生产者
1. 引入依赖
1
2
3
4
|
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
|
2. 定义配置文件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
spring:
cloud:
stream:
binders:
test:
type: rabbit
environment:
spring:
rabbitmq:
addresses: 10.0.20.132
port: 5672
username: root
password: root
virtual-host: /unicode-pay
bindings:
testOutPut:
destination: testRabbit
content-type: application/json
default-binder: test
|
现在来解释一下这些配置的含义
- binders： 这是一组binder的集合，这里配置了一个名为test的binder，这个binder中是包含了一个rabbit的连接信息
- bindings：这是一组binding的集合，这里配置了一个名为testOutPut的binding，这个binding中配置了指向名test的binder下的一个交换机testRabbit。
- 扩展： 如果我们项目中不仅集成了rabbit还集成了kafka那么就可以新增一个类型为kafka的binder、如果项目中会使用多个交换机那么就使用多个binding，
3.创建通道
1
2
3
4
5
6
7
8
|
public interface MqMessageSource {
String TEST_OUT_PUT = "testOutPut";
@Output(TEST_OUT_PUT)
MessageChannel testOutPut();
}
|
这个通道的名字就是上方binding的名字
4. 发送消息
1
2
3
4
5
6
7
8
9
10
11
12
|
@EnableBinding(MqMessageSource.class)
public class MqMessageProducer {
@Autowired
@Output(MqMessageSource.TEST_OUT_PUT)
private MessageChannel channel;
public void sendMsg(String msg) {
channel.send(MessageBuilder.withPayload(msg).build());
System.err.println("消息发送成功："+msg);
}
}
|
这里就是使用上方的通道来发送到指定的交换机了。需要注意的是withPayload方法你可以传入任何类型的对象，但是需要实现序列化接口
5. 创建测试接口
EnableBinding注解绑定的类默认是被Spring管理的，我们可以在controller中注入它
1
2
3
4
5
6
7
8
|
@Autowired
private MqMessageProducer mqMessageProducer;
@GetMapping(value = "/testMq")
public String testMq(@RequestParam("msg")String msg){
mqMessageProducer.sendMsg(msg);
return "发送成功";
}
|
生产者的代码到此已经完成了。
创建消费者
1. 引入依赖
1
2
3
4
|
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
|
2. 定义配置文件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
spring:
cloud:
stream:
binders:
test:
type: rabbit
environment:
spring:
rabbitmq:
addresses: 10.0.20.132
port: 5672
username: root
password: root
virtual-host: /unicode-pay
bindings:
testInPut:
destination: testRabbit
content-type: application/json
default-binder: test
|
这里与生产者唯一不同的地方就是testIntPut了，相信你已经明白了，它是binding的名字，也是通道与交换机绑定的关键
3.创建通道
1
2
3
4
5
6
7
8
|
public interface MqMessageSource {
String TEST_IN_PUT = "testInPut";
@Input(TEST_IN_PUT)
SubscribableChannel testInPut();
}
|
4. 接受消息
1
2
3
4
5
6
7
8
|
@EnableBinding(MqMessageSource.class)
public class MqMessageConsumer {
@StreamListener(MqMessageSource.TEST_IN_PUT)
public void messageInPut(Message<String> message) {
System.err.println(" 消息接收成功：" + message.getPayload());
}
}
|
这个时候启动Eureka、消息生产者和消费者，然后调用生产者的接口应该就可以接受到来自mq的消息了。
GitHub地址：https://github.com/shiyujun/spring-cloud-demo。代码所在模块：cloud-demo-consumer，cloud-demo-provider-2
如果对您有所帮助，请记得帮忙点一个star哦
本文出自http://zhixiang.org.cn,转载请保留