【SpringCloud】SpringCloud Stream訊息驅動

阿新 • • 發佈：2020-08-23

SpringCloud Stream訊息驅動

訊息驅動概述

是什麼

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

應用程式通過inputs或者outputs與Spring Cloud Stream中binder物件互動。
通過我們配置來binding(繫結)，而Spring Cloud Stream的binder物件負責與訊息中介軟體互動。
所以,我們只需要搞清楚如何與Spring Cloud Stream父互就可以方便使用訊息驅動的方式。

通過使用Spring Integration來連線訊息代理中介軟體以實現訊息事件驅動。

Spring Cloud Stream為-些供應商的訊息中介軟體產品提供了個性化的自動化配置實現，引用了釋出訂閱、消費組、分割槽的三個核心概念。

目前僅支援RabbitMQ、Kafka。

一句話

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

官網

https://spring.io/projects/spring-cloud-stream

Spring Cloud Stream中文指導手冊
https://blog.csdn.net/qq_32734365/article/details/81413218#spring-cloud-stream%E4%B8%AD%E6%96%87%E6%8C%87%E5%AF%BC%E6%89%8B%E5%86%8C

設計思想

標準MQ

生產者/消費者之間靠訊息媒介傳遞資訊內容

Message

訊息必須走特定的通道

訊息通道MessageChannel

訊息通道里的訊息如何被消費呢，誰負責收發處理

訊息通道MessageChannel的子介面SubscribableChannel，由MessageHandler訊息處理器所訂閱

為什麼使用Cloud Stream

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

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

stream憑什麼可以統一底層差異

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

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

Binder

在沒有繫結器這個概念的情況下，我們的SpringBoot應用要直接與訊息中介軟體進行資訊互動的時候,由於各訊息中介軟體構建的初衷不同，它們的實現細節上會有較大的差異性.通過定義繫結器作為中間層，完美地實現了應用程式與訊息中介軟體細節之間的隔離。Stream對消息中介軟體的進一步封裝可以做到程式碼層面對中介軟體的無感知，甚於動態的切換中介軟體(rabbitmq切換為kafka), 使得微服務開發的高度解耦，服務可以關注更多自己的業務流程。

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

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

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

Topic主題進行廣播

在RabbitMQ就是Exchange
在Kafka中就是Topic

Spring Cloud Stream標準流程套路

Binder

很方便的連線中介軟體，遮蔽差異

Channel

通道，是佇列Queue的一種抽象，在訊息通訊系統中就是實現儲存和轉發的媒介，通過Channel對佇列進行配置

Source和Sink

簡單的可以理解為參照物件是Spring Cloud Stream 自身，從Stream釋出訊息就是輸出，接受訊息就是輸入

編碼API和常用註解

案例說明

RabbitMQ環境已經OK

工程中新建三個子模組

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

訊息驅動之生產者

新建Module

cloud-stream-rabbitmq-provider8801

POM

<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>
    <!--eureka client-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
    <!--stream rabbit -->
    <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>

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:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 #設定心跳的時間間隔（預設是30S)
    lease-expiration-duration-in-seconds: 5 #如果超過5S間隔就登出節點 預設是90s
    instance-id: send-8801.com #在資訊列表時顯示主機名稱
    prefer-ip-address: true #訪問的路徑變為IP地址

主啟動類StreamMQMain8801

@SpringBootApplication
public class StreamMQMain8801 {

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

業務類

傳送訊息介面

public interface IMessageProvider {

    public String send();
}

傳送訊息介面實現類

package com.eiletxie.springcloud.service.impl;

import com.eiletxie.springcloud.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;

/**
 * @Author EiletXie
 * @Since 2020/3/14 14:13
 */
@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;
    }
}

Controller

@RestController
public class SendMessageController {

    @Resource
    private IMessageProvider messageProvider;

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

測試

啟動7001eureka

啟動rabbitmq

rabbitmq-plugins enbale rabbitmq_management
http://localhost:15672/

啟動8801

訪問：http://localhost:8801/sendMessage

訊息驅動之消費者

新建Module

cloud-stream-rabbitmq-consumer8802

POM

<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>
    <!--eureka client-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
    <!--stream rabbit -->
    <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>

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:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2 #設定心跳的時間間隔（預設是30S)
    lease-expiration-duration-in-seconds: 5 #如果超過5S間隔就登出節點 預設是90s
    instance-id: receive-8802.com #在資訊列表時顯示主機名稱
    prefer-ip-address: true #訪問的路徑變為IP地址

主啟動類

@SpringBootApplication
public class StreamMQMain8802 {

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

業務類

@Component
@EnableBinding(Sink.class)
public class ReceiverMessageListenerController {

    @Value("${server.port}")
    private  String serverPort;

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

分組消費與持久化

依照8802，clone出來一份8803

啟動

RabbitMQ

7001服務註冊

8801訊息生產

8802訊息消費

8803訊息消費

執行後有兩個問題

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

消費

目前是8802/8803同時收到了，存在重複消費問題

http://localhost:8801/sendMessage

如何解決
分組和持久化屬性group（重要）

生產實際案例

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

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

分組

原理

微服務應用放置於同一個group中，就能夠保證訊息只會被其中一個應用消費一次。不同的組是可以消費的，同一個組內會發生競爭關係，只有其中一個可以消費

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

group：atguiguA，atguiguB

8802修改YML

bindings: #服務的整合處理
  input: #這個名字是一個通道的名稱
    destination: studyExchange #表示要使用的Exchange名稱定義
    content-type: application/json #設定訊息型別，本次為json，本文要設定為“text/plain”
    binder: defaultRabbit #設定要繫結的訊息服務的具體設定
    group: atguiguA

在bindings下新增group屬性即可

8803修改YML

bindings: #服務的整合處理
  input: #這個名字是一個通道的名稱
    destination: studyExchange #表示要使用的Exchange名稱定義
    content-type: application/json #設定訊息型別，本次為json，本文要設定為“text/plain”
    binder: defaultRabbit #設定要繫結的訊息服務的具體設定
    group: atguiguB

我們自己配置

分散式微服務應用為了實現高可用和負載均衡，實際上都會部署多個例項，本例我啟動了兩個消費微服務(8802/8803)

多數情況，生產者傳送訊息給某個具體微服務時只希望被消費一次, 按照上面我們啟動兩個應用的例子，雖然它們同屬一個應用,但是這個訊息出現了被重複消費兩次的情況。為了解決這個問題，在Spring Cloud Stream中提供了消費組的概念。

結論

8802/8803實現了輪詢分組，每次只有一個消費者，8801模組的發的訊息只能被8802或8803其中一個接收到，這樣避免了重複消費

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

group：atguiguA

8802修改YML

8803修改YML

結論

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

持久化

通過上述，解決了重複消費問題，再看看持久化
停止8802/8803並去除掉8802分組group：atguiguA
8803的分組group：atguiguA沒有去掉
8801先發送4條訊息到rabbitmq，再試著啟動8802、8803看他們是否能接收到傳送過的訊息
先啟動8802，無分組屬性配置，後臺沒有打出來訊息
再啟動8803，有分組屬性配置，後臺打出來了MQ上的訊息

我們發現有指定分組的服務8803，訊息可以持久化，即使服務中途斷開後重啟仍然可以獲得，而未指定分組的服務就會丟失斷開期間傳送到MQ的訊息。

理解：因為訊息傳送時，會將訊息發給各個分組，雖然8803程式已經停止了，但是它的分組B此時還是存在的，所以訊息直接發給了分組B；（其實分組A也能收到訊息）
而8802程式此時沒有分組，也就是說，當8802啟動時才給它分配預設分組，而此時新建的預設分組是在傳送訊息之後新建的，所以錯過了此訊息。

【SpringCloud】SpringCloud Stream訊息驅動

訊息驅動概述

是什麼

一句話

官網

設計思想

標準MQ

生產者/消費者之間靠訊息媒介傳遞資訊內容

訊息必須走特定的通道

訊息通道里的訊息如何被消費呢，誰負責收發處理

為什麼使用Cloud Stream

stream憑什麼可以統一底層差異

Binder

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

Topic主題進行廣播

Spring Cloud Stream標準流程套路

Binder

Channel

Source和Sink

編碼API和常用註解

案例說明

RabbitMQ環境已經OK

工程中新建三個子模組

訊息驅動之生產者

新建Module

POM

YML

主啟動類StreamMQMain8801

業務類

傳送訊息介面

傳送訊息介面實現類

Controller

測試

訊息驅動之消費者

新建Module

POM

YML

主啟動類

業務類

分組消費與持久化

依照8802，clone出來一份8803

啟動

RabbitMQ

7001服務註冊

8801訊息生產

8802訊息消費

8803訊息消費

執行後有兩個問題

消費

生產實際案例

分組

原理

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

group：atguiguA，atguiguB

8802修改YML

8803修改YML

我們自己配置

結論

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

group：atguiguA

8802修改YML

8803修改YML

結論

持久化

相關推薦