SpringCloud 2020.0.4 系列之 Stream 延遲消息 的實現
- 2021 年 11 月 15 日
- 筆記
- JAVA, RabbitMQ, springboot, SpringCloud, Stream, 函數式編程, 分佈式, 微服務, 消息隊列
1. 概述
老話說的好：對待工作要有責任心，不僅要完成自己的部分，還要定期了解整體的進展。
言歸正傳，我們在開發產品時，常常會遇到一段時間後檢查狀態的場景，例如：用戶下單場景，如果訂單生成30分鐘後，用戶還沒有完成支付，則系統自動將訂單關閉。
在沒有消息中間件之前，常常是啟動一個定時程序，固定間隔的去檢查，不僅耗費系統資源，還會有較大的時間誤差。
今天我們就來聊一下 RabbitMQ 的 延遲消息 功能的使用。
RabbitMQ 鏡像模式集群的搭建，可參見我的另一篇文章《RabbitMQ 3.9.7 鏡像模式集群的搭建》（//www.cnblogs.com/w84422/p/15356202.html）
在早期的 SpringCloud 版本中常使用 @Input、@Output、@EnableBinding 和 @StreamListener 註解開發生產者與消費者。
官方原文：Deprecated as of 3.1 in favor of functional programming model。
SpringCloud 2020.0.4 版本中，已經不推薦這麼開發了，因此這裡我們也使用新的寫法（函數式編程方式） 開發。
閑話不多說，直接上代碼。
2. 延遲消息插件的安裝
2.1 下載插件
可以到 RabbitMQ 的官網下載，選擇與安裝的 RabbitMQ 對應的版本
官網地址：//www.rabbitmq.com/community-plugins.html
2.2 解壓插件
將插件 .ez 後綴名改為 .zip，然後使用 unzip 命令解壓。
2.3 拷貝插件到 RabbitMQ 的 plugins 文件夾下
如果是用 rpm 包的方式安裝的 RabbitMQ，路徑為： /usr/lib/rabbitmq/lib/rabbitmq_server-3.9.7/plugins
如果是 Docker 安裝，則 Docker 容器內的路徑為：/plugins
2.4 啟用延遲消息插件
# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
2.5 重啟 RabbitMQ
# /sbin/service rabbitmq-server stop
# /sbin/service rabbitmq-server start
3. 延遲消息發送DEMO
3.1 主要依賴
<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-stream-rabbit</artifactId>
</dependency>
3.2 消息實體類
@Setter @Getter public class MyMessage implements java.io.Serializable { // 消息體 private String payload; }
3.3 生產者
// 發送延遲消息 @PostMapping("/delayed") public String sendDelayedMessage(@RequestParam("body") String body, @RequestParam("seconds") Integer seconds) { MyMessage myMessage = new MyMessage(); myMessage.setPayload(body); // 生產消息 // 第一個參數是綁定名稱，格式為：自定義的綁定名稱-out-0，myDelayed是自定義的綁定名稱，out代表生產者，0是固定寫法 // 自定義的綁定名稱必須與消費方法的方法名保持一致 // 第二個參數是發送的消息實體 streamBridge.send("myDelayed-out-0", MessageBuilder.withPayload(myMessage) .setHeader("x-delay", seconds * 1000) .build() ); log.info("發送延遲消息成功"); return "SUCCESS"; }
3.4 消費者
// 消費延遲消息 @Bean public Consumer<Message<MyMessage>> myDelayed() { // 方法名必須與生產消息時自定義的綁定名稱一致 return message -> { log.info("接收延遲消息：{}", message.getPayload().getPayload()); }; }
3.5 application.yml 配置
spring: application: name: my-stream-new rabbitmq: # RabbitMQ 配置 addresses: 192.168.1.12:5672 username: guest password: guest virtual-host: / connection-timeout: 16000 cloud: function: # 定義消費者，多個用分號分隔，當存在大於1個的消費者時，不定義不會生效 definition: myDelayed stream: bindings: # 延遲消息 myDelayed-in-0: # 消費者綁定名稱，myDelayed是自定義的綁定名稱，in代表消費者，0是固定寫法 destination: my-delayed-topic # 對應的真實的 RabbitMQ Exchange myDelayed-out-0: # 生產者綁定名稱，myDelayed是自定義的綁定名稱，out代表生產者，0是固定寫法 destination: my-delayed-topic # 對應的真實的 RabbitMQ Exchange rabbit: bindings: myDelayed-in-0: consumer: delayedExchange: true # 開啟延遲功能 myDelayed-out-0: producer: delayedExchange: true # 開啟延遲功能
3.6 驗證延遲消息
發送延遲消息接口：
POST //localhost:49000/stream/delayed?body=這是一條延遲消息！&seconds=10
自動生成的 Exchange
自動生成的 Queue
消費情況
4. 綜述
今天聊了一下 SpringCloud Stream 組件 延遲消息 的實現 ，希望可以對大家的工作有所幫助。
歡迎幫忙點贊、評論、轉發、加關注 ：）
關注追風人聊Java，每天更新Java乾貨。
5. 個人公眾號
追風人聊Java，歡迎大家關注
