小米官宣：2023年上半年出貨量中國第一！
今日早間，小米電視官方微博帶來消息，稱2023年小米電視上半年出貨量達到了中國第一，同時還表示小米電視的巨屏風暴即將開始。“公布一個好消息2023年#小米電視上半年出貨量中國
哈嘍，大家好，我是了不起。
在實際的企業開發中，消息中間件是至關重要的組件之一。如常見的RabbitMQ和Kafka，這些中間件的差異性導致我們實際項目開發給我們造成了一定的困擾，這時候 Spring Cloud Stream 給我們提供了一種解耦合的方式。
Spring Cloud Stream 由一個中間件中立的核組成。
應用通過 Spring Cloud Stream 插入的Input(相當于消費者Consumer，它是從隊列中接收消息的)和Output(相當于生產者Producer，它是從隊列中發送消息的。)通道與外界交流。
通道通過指定中間件的Binder實現與外部代理連接。
業務開發者不再關注具體消息中間件，只需關注Binder對應用程序提供的抽象概念來使用消息中間件實現業務即可。
Spring Cloud Stream 為各大消息中間件產品提供了個性化的自動化配置實現，引用了發布-訂閱、消費組、分區的三個核心概念。
Spring Cloud Stream 提供了很多抽象和基礎組件來簡化消息驅動型微服務應用。包含以下內容：
Spring Cloud Stream由一個中立的中間件內核組成。Spring Cloud Stream會注入輸入和輸出的channels，應用程序通過這些channels與外界通信，而channels則是通過一個明確的中間件Binder與外部brokers連接。
圖片
Spring Cloud Stream 提供對Kafka、Rabbit MQ、Redis、Gemfire的Binder實現。Spring Cloud Stream還包括了一個TestSupportBinder、TestSupportBinder預留一個未更改的channel以便于直接地、可靠地和channels通信。
分區在有狀態處理中是一個很重要的概念，其重要性體現在性能和一致性上，要確保所有相關數據被一并處理，例如，在時間窗平均計算的例子中，給定傳感器測量結果應該都由同一應用實例進行計算。
Spring Cloud Stream支持在一個應用程序的多個實例之間數據分區，在分區的情況下，物理通信介質（例如，topic代理）被視為多分區結構。一個或多個生產者應用程序實例將數據發送給多個消費應用實例，并保證共同的特性的數據由相同的消費者實例處理。
Spring Cloud Stream 提供了一個通用的抽象，用于統一方式進行分區處理，因此分區可以用于自帶分區的代理（如Kafka）或者不帶分區的代理（如RabbieMQ）
Spring Cloud Stream 提供了一些預定義的注解，用于綁定輸入和輸出channels，以及如何監聽channels。
將@EnableBinding注解添加到應用的配置類，就可以把一個spring應用轉換成Spring Cloud Stream應用，@EnableBinding注解本身就包含@Configuration注解，會觸發Spring Cloud Stream 基本配置。
@Import(...)@Configuration@EnableIntegrationpublic @interface EnableBinding { ... Class<?>[] value() default {};}一個Spring Cloud Stream應用可以有任意數目的input和output通道，后者通過@Input和@Output注解在接口中定義。
定義在方法中，被修飾的方法注冊為消息中間件上數據流的事件監聽器，注解中屬性值對應了監聽的消息通道名。
Spring Cloud Stream提供了三個開箱即用的預定義接口。
public interface Source { String OUTPUT = "output"; @Output(Source.OUTPUT) MessageChannel output();}public interface Sink { String INPUT = "input"; @Input(Sink.INPUT) SubscribableChannel input();}public interface Processor extends Source, Sink {}下面是一個非常簡單的 SpringBootApplication應用，通過依賴Spring Cloud Stream，從Input通道監聽消息然后返回應答到Output通道，只要添加配置文件就可以應用。
@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())); }}下面解釋下這個示例中相關注解的應用：
消息發送失敗后悔發送到默認的一個“topic.errors"的channel中(topic是配置的destination)。要配置消息發送失敗的處理，需要將錯誤消息的channel打開。
消費者配置如下
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信息
@Bean("stream-test-topic.errors")MessageChannel testoutPutErrorChannel(){ return new PublishSubscribeChannel();}新建異常處理service
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); }}當發生異常時，由于測試類中已經將異常捕獲，處理發送異常主要是在這里進行。
這篇文章根據 Spring Cloud Stream 的官方文檔，對Stream做了一個整體的介紹，包括設計目標，應用場景，業務模型以及對外開放的注解，希望大家能夠學以致用。
本文鏈接：http://www.www897cc.com/showinfo-26-66963-0.htmlStream幫你無感知切換消息中間件
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
上一篇： Python中最常用的十個內置函數！
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號