Springboot詳解RocketMQ實現廣播消息流程
RocketMQ消息模式主要有兩種:廣播模式、集群模式(負載均衡模式)
廣播模式是每個消費者,都會消費消息;
負載均衡模式是每一個消費隻會被某一個消費者消費一次;
我們業務上一般用的是負載均衡模式,當然一些特殊場景需要用到廣播模式,比如發送一個信息到郵箱,手機,站內提示;
我們可以通過@RocketMQMessageListener
的messageModel
屬性值來設置,MessageModel.BROADCASTING
是廣播模式,MessageModel.CLUSTERING
是默認集群負載均衡模式
下面來介紹下 springboot+rockermq 整合實現 廣播消息
- 創建Springboot項目,添加rockermq 依賴
<!--rocketMq依賴--> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.2.1</version> </dependency>
- 配置rocketmq
# 端口
server:
port: 8083# 配置 rocketmq
rocketmq:
name-server: 127.0.0.1:9876
#生產者
producer:
#生產者組名,規定在一個應用裡面必須唯一
group: group1
#消息發送的超時時間 默認3000ms
send-message-timeout: 3000
#消息達到4096字節的時候,消息就會被壓縮。默認 4096
compress-message-body-threshold: 4096
#最大的消息限制,默認為128K
max-message-size: 4194304
#同步消息發送失敗重試次數
retry-times-when-send-failed: 3
#在內部發送失敗時是否重試其他代理,這個參數在有多個broker時才生效
retry-next-server: true
#異步消息發送失敗重試的次數
retry-times-when-send-async-failed: 3
- 生產端:新建一個 controller 來做消息發送
生產端按正常發送邏輯發送消息即可
package com.example.springbootrocketdemo.controller; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; /** * 廣播消息 * @author qzz */ @RestController public class RocketMQBroadCOntroller { @Autowired private RocketMQTemplate rocketMQTemplate; /** * 發送廣播消息 */ @RequestMapping("/testBroadSend") public void testSyncSend(){ //參數一:topic 如果想添加tag,可以使用"topic:tag"的寫法 //參數二:消息內容 for(int i=0;i<10;i++){ rocketMQTemplate.convertAndSend("test-topic-broad","test-message"+i); } } }
- 創建兩個消費者來消費消息
我們先集群負載均衡測試,加上messageModel=MessageModel.CLUSTERING
消費者1:
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 廣播消息 * 配置RocketMQ監聽 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:廣播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.CLUSTERING) public class RocketMQBroadConsumerListener implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("集群模式 消費者1,消費消息:"+s); } }
消費者2: 與消費者1在 同一個consumerGroup 和 topic
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 廣播消息 * 配置RocketMQ監聽 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:廣播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.CLUSTERING) public class RocketMQBroadConsumerListener2 implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("集群模式 消費者2,消費消息:"+s); } }
- 啟動服務,測試 集群模式消費
集群模式測試: 兩個消費者平攤 消息
- 把上面兩個消費者的 messageModel 屬性值修改成 廣播模式
消費者1:
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 廣播消息 * 配置RocketMQ監聽 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:廣播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.BROADCASTING) public class RocketMQBroadConsumerListener implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("廣播消息1 廣播模式,消費消息:"+s); } }
消費者2: 與消費者1在 同一個consumerGroup 和 topic
package com.example.springbootrocketdemo.config; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; /** * 廣播消息 * 配置RocketMQ監聽 * MessageModel.CLUSTERING:集群模式 * MessageModel.BROADCASTING:廣播模式 * @author qzz */ @Service @RocketMQMessageListener(consumerGroup = "test-broad",topic = "test-topic-broad",messageModel = MessageModel.BROADCASTING) public class RocketMQBroadConsumerListener2 implements RocketMQListener<String> { @Override public void onMessage(String s) { System.out.println("廣播消息2 廣播模式,消費消息:"+s); } }
- 重啟服務,測試 廣播模式消費
廣播模式消費下,兩個消費者都消費到Topic的所有消息。
測試成功!
到此這篇關於Springboot詳解RocketMQ實現廣播消息流程的文章就介紹到這瞭,更多相關Springboot廣播消息內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Springboot 整合 RocketMQ 收發消息的配置過程
- SpringBoot整合RocketMQ實現消息發送和接收的詳細步驟
- 解決SpringBoot整合RocketMQ遇到的坑
- RocketMQTemplate 註入失敗的解決
- 解決springboot集成rocketmq關於tag的坑