整合Spring Cloud Stream Binder與RabbitMQ進行消息發送與接收

1 前言

Spring Cloud Stream專門用于事件驅動的微服務系統，使用消息中間件來收發資訊。使用Spring Cloud Stream可專注于業務開發，而不用花太多心思在應用與MQ之間的互動上。而且，在切換MQ後，也無須做太多的代碼改動。

本文将整合Spring Cloud Stream和RabbitMQ，以實作消息收發。

2 整合過程

2.1 添加依賴

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

不同的MQ使用不同的依賴，非常容易切換。

2.2 定義處理收發的方法

隊列無非就是收和發，是以我們要先定義好，怎麼樣發，怎麼樣收。

發送消息：

@Bean
public Supplier<String> pkslowSource() {
  return () -> {
    String message = "www.pkslow.com";
    log.info("Sending value: " + message);
    return message;
  };
}
      

隻發送一個String，一般業務通常為Entity類。這裡發送的内容也固定不變，實際業務可以通過查資料庫，讀檔案等方式擷取資料源。

接收消息：

@Bean
public Consumer<String> pkslowSink() {
  return message -> {
    log.info("Received message " + message);
  };
}
      

直接列印消息即可，項目中的邏輯可按具體業務實作。

2.3 配置屬性

配置RabbitMQ：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: pkslow
    password: 123456
      

配置Spring Cloud Stream的相關項：

spring:
  cloud:
    stream:
      function:
        definition: pkslowSource;pkslowSink
      bindings:
        pkslowSource-out-0:
         destination: pkslow-topic
        pkslowSink-in-0:
          destination: pkslow-topic
      poller:
        fixed-delay: 500
      

spring.cloud.stream.function.definition會定義處理方法，如本文的收發消息的方法；

bindings配置對應的function；destination指向MQ的主題；

這裡配了一個poller，每隔500ms就會發送一次消息。

2.4 運作

先啟動個RabbitMQ：

docker run \
-e RABBITMQ_DEFAULT_USER=pkslow \
-e RABBITMQ_DEFAULT_PASS=123456 \
-p 5672:5672 -p 15672:15672 \
rabbitmq:3.8-management
      

運作程式後，會自己建立主題、發送資訊、接收資訊：

運作日志如下：