（十六）、SpringCloud Stream訊息驅動

阿新 • • 發佈：2021-10-01

Stream為什麼被引入

常見MQ(訊息中介軟體)：

ActiveMQ
RabbitMQ
RocketMQ
Kafka

有沒有一種新的技術誕生，讓我們不再關注具體MQ的細節，我們只需要用一種適配繫結的方式，自動的給我們在各種MQ內切換。（類似於Hibernate）

Cloud Stream是什麼？遮蔽底層訊息中介軟體的差異，降低切換成本，統一訊息的程式設計模型。

Stream是什麼及Binder介紹

官方文件1

官方文件2

Cloud Stream中文指導手冊

什麼是Spring Cloud Stream？

官方定義Spring Cloud Stream是一個構建訊息驅動微服務的框架。

應用程式通過inputs或者 outputs 來與Spring Cloud Stream中binder物件互動。

通過我們配置來binding(繫結)，而Spring Cloud Stream 的binder物件負責與訊息中介軟體互動。所以，我們只需要搞清楚如何與Spring Cloud Stream互動就可以方便使用訊息驅動的方式。通過使用Spring Integration來連線訊息代理中介軟體以實現訊息事件驅動。
Spring Cloud Stream為一些供應商的訊息中介軟體產品提供了個性化的自動化配置實現，引用了釋出-訂閱、消費組、分割槽的三個核心概念。

目前僅支援RabbitMQ、 Kafka。

Stream的設計思想

標準MQ

生產者/消費者之間靠訊息媒介傳遞資訊內容
訊息必須走特定的通道 - 訊息通道 Message Channel

訊息通道里的訊息如何被消費呢，誰負責收發處理 - 訊息通道MessageChannel的子介面SubscribableChannel，由MessageHandler訊息處理器所訂閱。

為什麼用Cloud Stream？

比方說我們用到了RabbitMQ和Kafka，由於這兩個訊息中介軟體的架構上的不同，像RabbitMQ有exchange，kafka有Topic和Partitions分割槽。

這些中介軟體的差異性導致我們實際專案開發給我們造成了一定的困擾，我們如果用了兩個訊息佇列的其中一種，後面的業務需求，我想往另外一種訊息佇列進行遷移，這時候無疑就是一個災難性的，一大堆東西都要重新推倒重新做，因為它跟我們的系統耦合了，這時候Spring Cloud Stream給我們提供了—種解耦合的方式。

Stream憑什麼可以統一底層差異？

在沒有繫結器這個概念的情況下，我們的SpringBoot應用要直接與訊息中介軟體進行資訊互動的時候，由於各訊息中介軟體構建的初衷不同，它們的實現細節上會有較大的差異性通過定義繫結器作為中間層，完美地實現了應用程式與訊息中介軟體細節之間的隔離。通過嚮應用程式暴露統一的Channel通道，使得應用程式不需要再考慮各種不同的訊息中介軟體實現。

通過定義繫結器Binder作為中間層，實現了應用程式與訊息中介軟體細節之間的隔離。

Binder：

INPUT對應於消費者
OUTPUT對應於生產者

Stream中的訊息通訊方式遵循了釋出-訂閱模式

Topic主題進行廣播

在RabbitMQ就是Exchange
在Kakfa中就是Topic

Stream編碼常用註解簡介

Binder - 很方便的連線中介軟體，遮蔽差異。
Channel - 通道，是佇列Queue的一種抽象，在訊息通訊系統中就是實現儲存和轉發的媒介，通過Channel對佇列進行配置。
Source和Sink - 簡單的可理解為參照物件是Spring Cloud Stream自身，從Stream釋出訊息就是輸出，接受訊息就是輸入。

編碼API和常用註解

案例說明

準備RabbitMQ環境

工程中新建三個子模組

cloud-stream-rabbitmq-provider8801，作為生產者進行發訊息模組
cloud-stream-rabbitmq-consumer8802，作為訊息接收模組
cloud-stream-rabbitmq-consumer8803，作為訊息接收模組

Stream訊息驅動之生產者

新建Module：cloud-stream-rabbitmq-provider8801

POM

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>cloud2021</artifactId>
        <groupId>com.ylc.cloud</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>cloud-stream-rabbitmq-provider8801</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <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-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <!--基礎配置-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

YML

server:
  port: 8801

spring:
  application:
    name: cloud-stream-provider
  cloud:
    stream:
      binders: # 在此處配置要繫結的rabbitmq的服務資訊；
        defaultRabbit: # 表示定義的名稱，用於於binding整合
          type: rabbit # 訊息元件型別
          environment: # 設定rabbitmq的相關的環境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服務的整合處理
        output: # 這個名字是一個通道的名稱
          destination: studyExchange # 表示要使用的Exchange名稱定義
          content-type: application/json # 設定訊息型別，本次為json，文字則設定“text/plain”
          binder: defaultRabbit # 設定要繫結的訊息服務的具體設定

eureka:
  client: # 客戶端進行Eureka註冊的配置
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 設定心跳的時間間隔（預設是30秒）
    lease-expiration-duration-in-seconds: 5 # 如果現在超過了5秒的間隔（預設是90秒）
    instance-id: send-8801.com  # 在資訊列表時顯示主機名稱
    prefer-ip-address: true     # 訪問的路徑變為IP地址

主啟動類

package com.ylc.cloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StreamMQMain8801 {
    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8801.class,args);
    }
}

業務類

1.傳送訊息介面

package com.ylc.cloud.service;

public interface IMessageProvider {
    public String send();
}

2.傳送訊息介面實現類

package com.ylc.cloud.service.Imp;

import com.ylc.cloud.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;

import javax.annotation.Resource;
import java.util.UUID;


@EnableBinding(Source.class) //定義訊息的推送管道
public class MessageProviderImpl implements IMessageProvider
{
    @Resource
    private MessageChannel output; // 訊息傳送管道

    @Override
    public String send()
    {
        String serial = UUID.randomUUID().toString();
        output.send(MessageBuilder.withPayload(serial).build());
        System.out.println("*****serial: "+serial);
        return null;
    }
}

3.Controller

package com.ylc.cloud.controller;

import com.ylc.cloud.service.IMessageProvider;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
public class SendMessageController
{
    @Resource
    private IMessageProvider messageProvider;

    @GetMapping(value = "/sendMessage")
    public String sendMessage() {
        return messageProvider.send();
    }

}

測試

啟動 7001eureka
啟動 RabpitMq（79_Bus之RabbitMQ環境配置）
- rabbitmq-plugins enable rabbitmq_management
- http://localhost:15672/
啟動 8801
訪問 - http://localhost:8801/sendMessage
- 後臺將列印serial: UUID字串

Stream訊息驅動之消費者

新建Module：cloud-stream-rabbitmq-consumer8802

POM

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>cloud2021</artifactId>
        <groupId>com.ylc.cloud</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>cloud-stream-rabbitmq-consumer8802</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--基礎配置-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

YML

server:
  port: 8802

spring:
  application:
    name: cloud-stream-consumer
  cloud:
    stream:
      binders: # 在此處配置要繫結的rabbitmq的服務資訊；
        defaultRabbit: # 表示定義的名稱，用於於binding整合
          type: rabbit # 訊息元件型別
          environment: # 設定rabbitmq的相關的環境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服務的整合處理
        input: # 這個名字是一個通道的名稱
          destination: studyExchange # 表示要使用的Exchange名稱定義
          content-type: application/json # 設定訊息型別，本次為物件json，如果是文字則設定“text/plain”
          binder: defaultRabbit # 設定要繫結的訊息服務的具體設定

eureka:
  client: # 客戶端進行Eureka註冊的配置
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 # 設定心跳的時間間隔（預設是30秒）
    lease-expiration-duration-in-seconds: 5 # 如果現在超過了5秒的間隔（預設是90秒）
    instance-id: receive-8802.com  # 在資訊列表時顯示主機名稱
    prefer-ip-address: true     # 訪問的路徑變為IP地址

主啟動類

package com.ylc.cloud;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StreamMQMain8802 {
    public static void main(String[] args) {
        SpringApplication.run(StreamMQMain8802.class,args);
    }
}

業務類

package com.ylc.cloud.controller;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;


@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController
{
    @Value("${server.port}")
    private String serverPort;


    @StreamListener(Sink.INPUT)
    public void input(Message<String> message)
    {
        System.out.println("消費者1號,----->接受到的訊息: "+message.getPayload()+"\t  port: "+serverPort);
    }
}

測試

啟動EurekaMain7001
啟動StreamMQMain8801
啟動StreamMQMain8802
8801傳送8802接收訊息

Stream之訊息重複消費

依照8802，克隆出來一份執行8803 - cloud-stream-rabbitmq-consumer8803。

啟動

RabbitMQ
服務註冊 - 8801
訊息生產 - 8801
訊息消費 - 8802
訊息消費 - 8802

執行後有兩個問題

有重複消費問題
訊息持久化問題

消費

http://localhost:8801/sendMessage
目前是8802/8803同時都收到了，存在重複消費問題
如何解決：分組和持久化屬性group（重要）

生產實際案例

比如在如下場景中，訂單系統我們做叢集部署，都會從RabbitMQ中獲取訂單資訊，那如果一個訂單同時被兩個服務獲取到，那麼就會造成資料錯誤，我們得避免這種情況。這時我們就可以使用Stream中的訊息分組來解決。

注意在Stream中處於同一個group中的多個消費者是競爭關係，就能夠保證訊息只會被其中一個應用消費一次。不同組是可以全面消費的(重複消費)。

只要分在同一個組就可以，就能夠保證訊息只會被其中一個應用消費一次。

Stream之group解決訊息重複消費

原理

微服務應用放置於同一個group中，就能夠保證訊息只會被其中一個應用消費一次。

不同的組是可以重複消費的，同一個組內會發生競爭關係，只有其中一個可以消費。

8802/8803都變成不同組，group兩個不同

修改8002、8003的YAL檔案：

增加分組，使他們在同一個組

結論：同一個組的多個微服務例項，每次只會有一個拿到

默認同一個佇列裡的消費者是競爭關係，誰的速度快誰搶到

Stream之訊息持久化

停止8802/8803並去除掉8802的分組group: A_Group，8803的分組group: A_Group沒有去掉。

8801先發送4條訊息到RabbitMq。

先啟動8802，無分組屬性配置，後臺沒有打出來訊息。

再啟動8803，有分組屬性配置，後臺打出來了MQ上的訊息。(訊息持久化體現)

原理是，exchange資料傳送到佇列中，由於02重啟沒有設定分組，會重新建立佇列並監聽，而03還是監聽原來佇列。

（十六）、SpringCloud Stream訊息驅動

Stream為什麼被引入

Stream是什麼及Binder介紹

Stream的設計思想

Stream編碼常用註解簡介

Stream訊息驅動之生產者

Stream訊息驅動之消費者

Stream之訊息重複消費

Stream之group解決訊息重複消費

Stream之訊息持久化

（十六）、SpringCloud Stream訊息驅動

五（十六）、介面

（十四）、SpringCloud Config分散式配置中心

（十七）、SpringCloud Sleuth分散式請求鏈路追蹤

（十九）、SpringCloud Alibaba Sentinel實現熔斷和限流

Java SE基礎鞏固（十六）：Stream（流）

SQLyog基礎操作（十六）-資料庫連線池（開源資料庫（DBCP、C3P0））

Dubbo原始碼解析（十六）遠端通訊——Netty3

Spark 系列（十六）—— Spark Streaming 整合 Kafka

Spring Boot入門系列（十六）使用pagehelper實現分頁功能

Quartz.Net系列（十六）：通過Plugins模式使用Xml方式配置Job和Trigger和自定義LogPrivider

Quartz.Net系列（十六）：Misfire策略在SimpleScheduler和CronScheduler中的使用

【016期】JavaSE面試題（十六）：反射

Android開發筆記（十六）——Activity的4種啟動模式

五（十一）、Object 類、Equals方法、toString方法、包裝類、Junit

資料結構與演算法（十六）：平衡二叉樹

深度學習與Pytorch入門實戰（十六）情感分類實戰（基於IMDB資料集）

設計模式學習筆記（十六）：職責鏈模式

Flink基礎（十六）：Table API 和 Flink SQL（五）函式（Functions）

五（十四）、final

（十六）、SpringCloud Stream訊息驅動

Stream為什麼被引入

Stream是什麼及Binder介紹

Stream的設計思想

Stream編碼常用註解簡介

Stream訊息驅動之生產者

Stream訊息驅動之消費者

Stream之訊息重複消費

Stream之group解決訊息重複消費

Stream之訊息持久化

相關推薦