日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

當前位置:首頁 > 科技  > 軟件

Stream幫你無感知切換消息中間件

來源: 責編: 時間:2024-01-24 09:02:56 227觀看
導讀哈嘍,大家好,我是了不起。在實際的企業開發中,消息中間件是至關重要的組件之一。如常見的RabbitMQ和Kafka,這些中間件的差異性導致我們實際項目開發給我們造成了一定的困擾,這時候 Spring Cloud Stream 給我們提供了一種解

哈嘍,大家好,我是了不起。U6228資訊網——每日最新資訊28at.com

在實際的企業開發中,消息中間件是至關重要的組件之一。如常見的RabbitMQ和Kafka,這些中間件的差異性導致我們實際項目開發給我們造成了一定的困擾,這時候 Spring Cloud Stream 給我們提供了一種解耦合的方式。U6228資訊網——每日最新資訊28at.com

簡介

Spring Cloud Stream 由一個中間件中立的核組成。U6228資訊網——每日最新資訊28at.com

應用通過 Spring Cloud Stream 插入的Input(相當于消費者Consumer,它是從隊列中接收消息的)和Output(相當于生產者Producer,它是從隊列中發送消息的。)通道與外界交流。U6228資訊網——每日最新資訊28at.com

通道通過指定中間件的Binder實現與外部代理連接。U6228資訊網——每日最新資訊28at.com

業務開發者不再關注具體消息中間件,只需關注Binder對應用程序提供的抽象概念來使用消息中間件實現業務即可。U6228資訊網——每日最新資訊28at.com

詳細介紹

核心概念

Spring Cloud Stream 為各大消息中間件產品提供了個性化的自動化配置實現,引用了發布-訂閱、消費組、分區的三個核心概念。U6228資訊網——每日最新資訊28at.com

Spring Cloud Stream 提供了很多抽象和基礎組件來簡化消息驅動型微服務應用。包含以下內容:U6228資訊網——每日最新資訊28at.com

  • Spring Cloud Stream的應用模型
  • 綁定抽象
  • 持久化發布/訂閱支持
  • 消費者組支持
  • 分片支持(Partitioning Support)
  • 可插拔API

應用模型

Spring Cloud Stream由一個中立的中間件內核組成。Spring Cloud Stream會注入輸入和輸出的channels,應用程序通過這些channels與外界通信,而channels則是通過一個明確的中間件Binder與外部brokers連接。U6228資訊網——每日最新資訊28at.com

圖片圖片U6228資訊網——每日最新資訊28at.com

各大消息中間件的綁定抽象

Spring Cloud Stream 提供對Kafka、Rabbit MQ、Redis、Gemfire的Binder實現。Spring Cloud Stream還包括了一個TestSupportBinder、TestSupportBinder預留一個未更改的channel以便于直接地、可靠地和channels通信。U6228資訊網——每日最新資訊28at.com

分區支持

分區在有狀態處理中是一個很重要的概念,其重要性體現在性能和一致性上,要確保所有相關數據被一并處理,例如,在時間窗平均計算的例子中,給定傳感器測量結果應該都由同一應用實例進行計算。U6228資訊網——每日最新資訊28at.com

Spring Cloud Stream支持在一個應用程序的多個實例之間數據分區,在分區的情況下,物理通信介質(例如,topic代理)被視為多分區結構。一個或多個生產者應用程序實例將數據發送給多個消費應用實例,并保證共同的特性的數據由相同的消費者實例處理。U6228資訊網——每日最新資訊28at.com

Spring Cloud Stream 提供了一個通用的抽象,用于統一方式進行分區處理,因此分區可以用于自帶分區的代理(如Kafka)或者不帶分區的代理(如RabbieMQ)U6228資訊網——每日最新資訊28at.com

編程模型

Spring Cloud Stream 提供了一些預定義的注解,用于綁定輸入和輸出channels,以及如何監聽channels。U6228資訊網——每日最新資訊28at.com

通過@EnableBinding觸發綁定

將@EnableBinding注解添加到應用的配置類,就可以把一個spring應用轉換成Spring Cloud Stream應用,@EnableBinding注解本身就包含@Configuration注解,會觸發Spring Cloud Stream 基本配置。U6228資訊網——每日最新資訊28at.com

@Import(...)@Configuration@EnableIntegrationpublic @interface EnableBinding {    ...    Class<?>[] value() default {};}

@Input 與 @Output

一個Spring Cloud Stream應用可以有任意數目的input和output通道,后者通過@Input和@Output注解在接口中定義。U6228資訊網——每日最新資訊28at.com

@StreamListener

定義在方法中,被修飾的方法注冊為消息中間件上數據流的事件監聽器,注解中屬性值對應了監聽的消息通道名。U6228資訊網——每日最新資訊28at.com

Source、Sink和Processor

Spring Cloud Stream提供了三個開箱即用的預定義接口。U6228資訊網——每日最新資訊28at.com

  • Source用于有單個輸出(outbound)通道的應用。
public interface Source {  String OUTPUT = "output";  @Output(Source.OUTPUT)  MessageChannel output();}
  • Sink用于有單個輸入(inbound)通道的應用。
public interface Sink {  String INPUT = "input";  @Input(Sink.INPUT)  SubscribableChannel input();}
  • Processor用于單個應用同時包含輸入和輸出通道的情況。
public interface Processor extends Source, Sink {}

極簡實例

下面是一個非常簡單的 SpringBootApplication應用,通過依賴Spring Cloud Stream,從Input通道監聽消息然后返回應答到Output通道,只要添加配置文件就可以應用。U6228資訊網——每日最新資訊28at.com

@SpringBootApplication@EnableBinding(Processor.class)public class ServiceApplication {    public static void main(String[] args) {        SpringApplication.run(MyLoggerServiceApplication.class, args);    }    @StreamListener(Processor.INPUT)    @SendTo(Processor.OUTPUT)    public LogMessage enrichLogMessage(LogMessage log) {        return new LogMessage(String.format("[1]: %s", log.getMessage()));    }}

下面解釋下這個示例中相關注解的應用:U6228資訊網——每日最新資訊28at.com

  • @EnableBinding聲明了這個應用程序綁定了2個通道:INPUT和OUTPUT。這2個通道是在接口Processor中定義的(Spring Cloud Stream默認設置)。所有通道都是配置在一個具體的消息中間件或綁定器中。
  • @StreamListener(Processor.INPUT)表明這里在input中提取消息,并且處理。
  • @SendTo(Processor.OUTPUT)表明在output中返回消息。

其他特性

消息發送失敗的處理

消息發送失敗后悔發送到默認的一個“topic.errors"的channel中(topic是配置的destination)。要配置消息發送失敗的處理,需要將錯誤消息的channel打開。U6228資訊網——每日最新資訊28at.com

消費者配置如下U6228資訊網——每日最新資訊28at.com

spring:  application:    name: spring-cloud-stream-producer  cloud:    stream:      rocketmq:        binder:          name-server: 127.0.0.1:9876        bindings:          output:            producer:              group: test              sync: true      bindings:        output:          destination: stream-test-topic          content-type: text/plain # 內容格式。這里使用 JSON          producer:            errorChannelEnabled: true

在啟動類中配置錯誤消息的Channel信息U6228資訊網——每日最新資訊28at.com

@Bean("stream-test-topic.errors")MessageChannel testoutPutErrorChannel(){    return new PublishSubscribeChannel();}

新建異常處理serviceU6228資訊網——每日最新資訊28at.com

import org.springframework.integration.annotation.ServiceActivator;import org.springframework.messaging.Message;import org.springframework.stereotype.Service;@Servicepublic class ErrorProducerService {    @ServiceActivator(inputChannel = "stream-test-topic.errors")    public void receiveProducerError(Message message){        System.out.println("receive error msg :"+message);    }}

當發生異常時,由于測試類中已經將異常捕獲,處理發送異常主要是在這里進行。U6228資訊網——每日最新資訊28at.com

總結

這篇文章根據 Spring Cloud Stream 的官方文檔,對Stream做了一個整體的介紹,包括設計目標,應用場景,業務模型以及對外開放的注解,希望大家能夠學以致用。U6228資訊網——每日最新資訊28at.com

本文鏈接:http://www.www897cc.com/showinfo-26-66963-0.htmlStream幫你無感知切換消息中間件

聲明:本網頁內容旨在傳播知識,若有侵權等問題請及時與本網聯系,我們將在第一時間刪除處理。郵件:2376512515@qq.com

上一篇: Python中最常用的十個內置函數!

下一篇: Ubuntu大佬神操作!Rust版Linux調度器秀麻了,性能遠超C!

標簽:
  • 熱門焦點
Top 主站蜘蛛池模板: 自贡市| 宝兴县| 临安市| 琼中| 陕西省| 开封市| 巴楚县| 汽车| 岱山县| 仲巴县| 宁陕县| 襄樊市| 舟山市| 农安县| 万源市| 舒城县| 正蓝旗| 巴马| 乌拉特中旗| 庄浪县| 明水县| 涞源县| 祁阳县| 休宁县| 皋兰县| 富裕县| 田东县| 长治市| 大姚县| 尤溪县| 北流市| 仲巴县| 奉节县| 平邑县| 阿图什市| 嵩明县| 图们市| 郎溪县| 阜康市| 桐乡市| 洛扎县|