天天學點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());
 }

}