天天学点SpringCloud（十三）：SpringCloud-Stream整合RabbitMQ

时间 2019-11-19

标签天天 springcloud 十三 stream 整合 rabbitmq 栏目 Apache 繁體版

原文原文链接

咱们知道，当微服务愈来愈来多的时候，仅仅是feign的http调用方式已经知足不了咱们的使用场景了。这个时候系统就须要接入消息中间件了。相比较于传统的Spring项目、SpringBoot项目使用消息中间件的不少配置不一样，SpringCloud Stream抽象了中间件产品的不一样，在SpringCloud中你仅仅须要修改几行配置文件就能够灵活的切换中间件产品而不须要修改任何代码。git

如今咱们以SpringCloud Stream整合RabbitMQ为例来学习一下github

建立生产者

1. 引入依赖

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

2. 定义配置文件

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

如今来解释一下这些配置的含义spring

binders：这是一组binder的集合，这里配置了一个名为test的binder，这个binder中是包含了一个rabbit的链接信息
bindings：这是一组binding的集合，这里配置了一个名为testOutPut的binding，这个binding中配置了指向名test的binder下的一个交换机testRabbit。
扩展：若是咱们项目中不只集成了rabbit还集成了kafka那么就能够新增一个类型为kafka的binder、若是项目中会使用多个交换机那么就使用多个binding，

3.建立通道

public interface MqMessageSource {

 String TEST_OUT_PUT = "testOutPut";

 @Output(TEST_OUT_PUT)
 MessageChannel testOutPut();

}

这个通道的名字就是上方binding的名字json

4. 发送消息

@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方法你能够传入任何类型的对象，可是须要实现序列化接口app

5. 建立测试接口

EnableBinding注解绑定的类默认是被Spring管理的，咱们能够在controller中注入它ide

@Autowired
private MqMessageProducer mqMessageProducer;

@GetMapping(value = "/testMq")
public String testMq(@RequestParam("msg")String msg){
 mqMessageProducer.sendMsg(msg);
 return "发送成功";
}

生产者的代码到此已经完成了。微服务

建立消费者

1. 引入依赖

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

2. 定义配置文件

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.建立通道

public interface MqMessageSource {

 String TEST_IN_PUT = "testInPut";

 @Input(TEST_IN_PUT)
 SubscribableChannel testInPut();

}

4. 接受消息

@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-2ui

若是对您有所帮助，请记得帮忙点一个star哦

本文出自http://zhixiang.org.cn,转载请保留