歡迎您光臨本站 註冊首頁

Spring Cloud Stream微服務消息框架原理及實例解析

←手機掃碼閱讀     ljg58026 @ 2020-06-03 , reply:0

隨著近些年微服務在國內的盛行,消息驅動被提到的越來越多。主要原因是系統被拆分成多個模塊後,一個業務往往需要在多個服務間相互調用,不管是採用HTTP還是RPC都是同步的,不可避免快等慢的情況發生,系統性能上很容易遇到瓶頸。在這樣的背景下,將業務中實時性要求不是特別高且非主幹的部分放到消息隊列中是很好的選擇,達到了異步解耦的效果。

目前消息隊列有很多優秀的中間件,目前使用較多的主要有 RabbitMQ,Kafka,RocketMQ 等,這些中間件各有優勢,有的對 AMQP(應用層標準高級消息隊列協議)支持完善,有的提供了更高的可靠性,有的對大數據支持良好,同時各種消息中間件概念不統一,使得選擇和使用一款合適的消息中間件成為難題。Spring跳出來給出瞭解決方案:Spring Cloud Stream,使用它可以很方便高效的操作消息中間件,程序員只要關心業務代碼即可,目前官方支持 RabbitMQ,Kafka兩大主流MQ,RocketMQ 則自己提供了相應支持。

首先看一下Spring Cloud Stream做了什麼,如下圖所示,框架目前官方把消息中間件抽象成了 Binder,業務代碼通過進出管道連接 Binder,各消息中間件的差異性統一交給了框架處理,程序員只需要瞭解框架的抽象出來的一些統一概念即可

  • Binder(綁定器):RabbitMQ,Kafka等中間件服務的封裝

  • Channel(管道):也就是圖中的 inputs 和 outputs 所指區域,是應用程序和 Binder 的橋樑

  • Gourp(消費組):由於微服務會部署多實例,為了保證只被服務的一個實例消費,可以通過配置,把實例都綁到同一個消費組

  • Partitioning (消息分區):如果某一類消息只想指定給服務的固定實例消費,可以使用分區實現

Spring Cloud Stream將業務代碼和消息中間件解耦,帶來的好處可以從下圖很直觀的感受到,很簡潔的代碼,我們便能從RabbitMQ中接受消息然後經過業務處理再向Kafka發送一條消息,只需要更改相關配置就能快速改變系統行為。

細心的讀者可能會好奇,上圖的代碼只是注入了一個簡單的 Function 而已,實際上,Spring Cloud Stream3.0後集成了Spring Cloud Function框架 ,提倡函數式的風格,棄用先前版本基於註解的開發方式。Spring Cloud Function是 Serverless 和 Faas 的產物,強調面向函數編程,一份代碼各雲平臺運行,和Spring Cloud Stream一樣也是解決了基礎設施的差異性問題,通過強大的自動裝配機制,可以根據配置自動暴露 HTTP 服務或者消息服務,並且同時支持命令式和響應式編程模式,可以說是很強大了。下面通過一個簡單的例子來理解下上圖的代碼和框架的使用把。

簡單案例

模擬一個簡單的下單,收到訂單之後處理完,返回成功,然後發送消息給庫存模塊,庫存模塊再發送消息給報表模塊

項目地址

springcloud-stream

項目結構

項目依賴

 org.springframework.bootspring-boot-starter-weborg.springframework.cloudspring-cloud-starter-stream-rabbit


表單

 @Data public class OrderForm { private String productName; }


消息管道註冊

 @Configuration @Slf4j public class MessageQueueConfig { @Bean public Functioninventory() { return orderForm -> { log.info("Inventory Received Message: " + orderForm); return orderForm; }; } @Bean public Consumerreport() { return orderForm -> { log.info("Report Received Message: " + orderForm); }; } }


Controller

 @Slf4j @RestController public class OrderController { @Autowired private BeanFactoryChannelResolver resolver; @PostMapping("order") public String order(@RequestBody OrderForm orderForm) { log.info("Received Request " + orderForm); resolver.resolveDestination("inventory-in-0").send(new GenericMessage<>(orderForm)); return "success"; } }


配置

框架會按照中間件默認端口去連接,這裡自定義了一個名為myLocalRabbit的類型是RabbitMQ的Binder配置,bindings下面 inventory-in-0 是通道名,接受inventory主題(對應RabbitMQ的ExChange)的消息,然後處理完通過 inventory-out-0 通道發送消息到 report 主題, report-in-0通道負責接受report主題的消息。

注:通道名=註冊的 function 方法名 + in或者out + 參數位置(詳見註釋)

 spring: cloud: stream: # 配置消息中間件信息 binders: myLocalRabbit: type: rabbit environment: spring: rabbitmq: host: localhost port: 31003 username: guest password: guest virtual-host: / # 重點,如何綁定通道,這裡有個約定,開頭是函數名,in表示消費消息,out表示生產消息,最後的數字是函數接受的參數的位置,destination後面為訂閱的主題 # 比如Function<Tuple2, Flux> gather() # gather函數接受的第一個String參數對應 gather-in-0,第二個Integer參數對應 gather-in-1,輸出對應 gather-out-0 bindings: inventory-in-0: destination: inventory inventory-out-0: destination: report report-in-0: destination: report # 註冊聲明的三個函數 function: definition: inventory;report


測試

 POST http://localhost:8080/order Content-Type: application/json { "productName": "999" }


結果

 POST http://localhost:8080/order HTTP/1.1 200 Content-Type: text/plain;charset=UTF-8 Content-Length: 7 Date: Sat, 30 May 2020 15:27:56 GMT Keep-Alive: timeout=60 Connection: keep-alive success Response code: 200; Time: 56ms; Content length: 7 bytes


後臺日誌

可以看到消息成功發送到了庫存和報表服務

 2020-05-30 23:27:56.956 INFO 8760 --- [nio-8080-exec-1] c.e.springcloudstream.OrderController : Received Request OrderForm(productName=999) 2020-05-30 23:27:56.956 INFO 8760 --- [nio-8080-exec-1] o.s.i.h.s.MessagingMethodInvokerHelper : Overriding default instance of MessageHandlerMethodFactory with provided one. 2020-05-30 23:27:56.957 INFO 8760 --- [nio-8080-exec-1] c.e.s.MessageQueueConfig : Inventory Received Message: OrderForm(productName=999) 2020-05-30 23:27:56.958 INFO 8760 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [localhost:31003] 2020-05-30 23:27:56.964 INFO 8760 --- [nio-8080-exec-1] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory.publisher#6131841e:0/SimpleConnection@192fe472 [delegate=amqp://guest@127.0.0.1:31003/, localPort= 2672] 2020-05-30 23:27:56.965 INFO 8760 --- [nio-8080-exec-1] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (inventory.anonymous.wtaFwHlNRkql5IUh2JCNAA) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost. 2020-05-30 23:27:56.965 INFO 8760 --- [nio-8080-exec-1] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (report.anonymous.SJgpJKiJQf2tudszgf623w) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost. 2020-05-30 23:27:56.979 INFO 8760 --- [f2tudszgf623w-1] o.s.i.h.s.MessagingMethodInvokerHelper : Overriding default instance of MessageHandlerMethodFactory with provided one. 2020-05-30 23:27:56.980 INFO 8760 --- [f2tudszgf623w-1] c.e.s.MessageQueueConfig : Report Received Message: OrderForm(productName=999)

[ljg58026 ] Spring Cloud Stream微服務消息框架原理及實例解析已經有341次圍觀

http://coctec.com/docs/java/show-post-236819.html