SpringCloud 2020.0.4 系列之 Stream 消息廣播與消息分組的實現

2021 年 11 月 12 日
筆記
JAVA, RabbitMQ, springboot, SpringCloud, Stream, 函數式編程, 分散式, 微服務, 消息隊列

1. 概述

老話說的好：事情太多，做不過來，就先把事情記在本子上，然後理清思路、排好優先順序，一件一件的去完成。

言歸正傳，今天我們來聊一下 SpringCloud 的 Stream 組件，SpringCloud Stream 是一個構建消息驅動微服務的框架，簡單理解就是在 SpringCloud 中操作消息中間件的組件。

這裡我們以 RabbitMQ 為例，介紹一下如何使用 Stream 實現消息廣播和消息分組。

RabbitMQ 鏡像模式集群的搭建，可參見我的另一篇文章《RabbitMQ 3.9.7 鏡像模式集群的搭建》（//www.cnblogs.com/w84422/p/15356202.html）

在早期的 SpringCloud 版本中常使用 @Input、@Output、@EnableBinding 和 @StreamListener 註解開發生產者與消費者。

官方原文：Deprecated as of 3.1 in favor of functional programming model。

SpringCloud 2020.0.4 版本中，已經不推薦這麼開發了，因此這裡我們也使用新的寫法（函數式編程方式）開發。

閑話不多說，直接上程式碼。

2. 消息廣播

2.1 主要依賴

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- 健康檢查 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

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

2.2 消息實體類

@Setter
@Getter
public class MyMessage implements java.io.Serializable {

    // 消息體
    private String payload;
}

2.3 生產者

    @Autowired
    private StreamBridge streamBridge;

    // 發送廣播消息
    @PostMapping("/broadcast")
    public String sendBroadcastMessage(@RequestParam("body") String body) {

        MyMessage myMessage = new MyMessage();
        myMessage.setPayload(body);

        // 生產消息
        // 第一個參數是綁定名稱，格式為：自定義的綁定名稱-out-0，myBroadcast是自定義的綁定名稱，out代表生產者，0是固定寫法
        // 自定義的綁定名稱必須與消費方法的方法名保持一致
        // 第二個參數是發送的消息實體
        streamBridge.send("myBroadcast-out-0", myMessage);
        return "SUCCESS";
    }

2.4 消費者

@Slf4j
@Component
// 消費者
public class Concumer {

    // 消費廣播消息
    @Bean
    public Consumer<MyMessage> myBroadcast() {  // 方法名必須與生產消息時自定義的綁定名稱一致

        return message -> {
            log.info("接收廣播消息：{}", message.getPayload());
        };
    }

2.5 application.yml 配置

spring:
  application:
    name: my-stream-new
  rabbitmq:   # RabbitMQ 配置
    addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000
  cloud:
    function:
      # 定義消費者，多個用分號分隔，當存在大於1個的消費者時，不定義不會生效
      definition: myBroadcast
    stream:
      bindings:
        
        # 廣播消息
        myBroadcast-in-0:   # 消費者綁定名稱，myBroadcast是自定義的綁定名稱，in代表消費者，0是固定寫法
          destination: my-broadcast-topic   # 對應的真實的 RabbitMQ Exchange
        myBroadcast-out-0:  # 生產者綁定名稱，myBroadcast是自定義的綁定名稱，out代表生產者，0是固定寫法
          destination: my-broadcast-topic   # 對應的真實的 RabbitMQ Exchange

注意：spring.cloud.function.definition 屬性非常重要，務必配置

2.6 驗證廣播消息

將 Stream 工程啟動兩個實例，調用發送廣播消息介面，查看消息消費情況。

發送廣播消息介面：

POST //localhost:49000/stream/broadcast?body=這是一條廣播消息！

自動生成的 Exchange

自動生成的 Queue

兩個實例的消費情況

3. 消息分組

消息分組後，同一分組的消費服務，只能有一個消費者消費到消息。

3.1 生產者

    // 發送分組消息
    @PostMapping("/group")
    public String sendGroupMessage(@RequestParam("body") String body) {

        MyMessage myMessage = new MyMessage();
        myMessage.setPayload(body);

        // 生產消息
        streamBridge.send("myGroup-out-0", myMessage);
        return "SUCCESS";
    }

3.2 消費者

    // 消費分組消息
    @Bean
    public Consumer<MyMessage> myGroup() {  // 方法名必須與生產消息時自定義的綁定名稱一致

        return message -> {
            log.info("接收分組消息：{}", message.getPayload());
        };
    }

3.3 application.yml 配置

spring:
  application:
    name: my-stream-new
  rabbitmq:   # RabbitMQ 配置
    addresses: 192.168.1.22:5672,192.168.1.12:5672,192.168.1.11:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000
  cloud:
    function:
      # 定義消費者，多個用分號分隔，當存在大於1個的消費者時，不定義不會生效
      definition: myBroadcast;myGroup
    stream:
      bindings:        
        # 廣播消息
        myBroadcast-in-0:   # 消費者綁定名稱，myBroadcast是自定義的綁定名稱，in代表消費者，0是固定寫法
          destination: my-broadcast-topic   # 對應的真實的 RabbitMQ Exchange
        myBroadcast-out-0:  # 生產者綁定名稱，myBroadcast是自定義的綁定名稱，out代表生產者，0是固定寫法
          destination: my-broadcast-topic   # 對應的真實的 RabbitMQ Exchange

        # 分組消息
        myGroup-in-0: # 消費者綁定名稱
          destination: my-group-topic   # 對應的真實的 RabbitMQ Exchange
          group: my-group-1   # 同一分組的消費服務，只能有一個消費者消費到消息
        myGroup-out-0: # 生產者綁定名稱
          destination: my-group-topic   # 對應的真實的 RabbitMQ Exchange