1. 程式人生 > >每天學點SpringCloud(十三):SpringCloud-Stream整合RabbitMQ

每天學點SpringCloud(十三):SpringCloud-Stream整合RabbitMQ

我們知道,當微服務越來越來多的時候,僅僅是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

現在來解釋一下這些配置的含義

  1. binders: 這是一組binder的集合,這裡配置了一個名為test的binder,這個binder中是包含了一個rabbit的連線資訊
  2. bindings:這是一組binding的集合,這裡配置了一個名為testOutPut的binding,這個binding中配置了指向名test的binder下的一個交換機testRabbit。
  3. 擴充套件: 如果我們專案中不僅集成了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/2388386839/spring-cloud-demo。程式碼所在模組:cloud-demo-consumer,cloud-demo-provider-2

如果對您有所幫助,請記得幫忙點一個star哦

本文出自http://zhixiang.org.cn,轉載請保留