SpringBoot整合RocketMQ的方法詳解
一:Ubuntu安裝RocketMQ
1.下載(在下面地址選擇自己需要的版本的rocketmq)
http://rocketmq.apache.org/release_notes/
2.解壓,更改配置
將下載的zip文件解壓到自己需要安裝的位置
在unbuntu系統下需要修改安裝跟目錄下的兩個文件
bin/runserver.sh 和 bin/runbroker.sh 文件
(1).都修改兩個文件的 java_home=/java安裝HOME路徑/
(2).都修改兩個文件的 java_home=/java安裝HOME路徑/jre/
(3).修改runserver.sh內存
JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m" # 原來是:JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn4g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
(4).修改runbroker.sh內存
JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g # 原來是:JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g
(5).創建日志目錄
mkdir logs
3.後臺掛載啟動 mqnamesrv(在安裝根目錄下執行)
nohup sh bin/mqnamesrv > logs/namesrv.log 2>&1 &
5.查看mqnamesrv是否啟動成功
查看 logs/namesrv.log 文件出現以下提示表示啟動成功!
The Name Server boot success. serializeType=JSON
6.安裝 mqbroker
Broker是 RocketMQ 的核心,大部分'重量級'工作都是由 Broker完成的,
包括接收 Producer 發過來的消息、處理 Consumer 的消費消息請求、消息的持 久化存儲、消息的 HA 機制以及服務端過濾功能等 。
(1).更改broker配置,在安裝根目錄下conf/broker.conf文件 添加一行
brokerIP1 = 127.0.0.0 # IP根據自己電腦服務器修改一下,最好不要使用這個,ifconfig就可以查看到自己服務器的IP
(2).掛載啟動 mqbroker(在安裝根目錄下)
nohup sh bin/mqbroker -c conf/broker.conf -n localhost:9876 > logs/broker.log 2>&1 &
(3).查看 mqbroker 是否啟動成功
查看 logs/broker.log 文件出現以下提示表示啟動成功!
The broker[broker-a, 127.0.0.0:10911] boot success. serializeType=JSON and name server is localhost:9876
7.關閉broker和nameserver(在安裝根目錄下執行)
./bin/mqshutdown broker ./bin/mqshutdown namesrv
8.創建一個topic
./bin/mqadmin updateTopic -n '127.0.0.0:9876' -c MyCluster -t MyTopic
9.查看集群狀態
./bin/mqadmin clusterList -n localhost:9876
二:添加RocketMQ依賴
<!--rocket mq 依賴--> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.1.1</version> <exclusions> <exclusion> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </exclusion> <exclusion> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> </exclusion> <exclusion> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> </exclusion> </exclusions> </dependency>
三:在application中添加RocketMQ配置
#rocketmq 配置 rocketmq: name-server: 127.0.0.0:9876 #IP根據自己電腦服務器配置的 mqbroker->brokerIP1 修改一下,最好不要使用這個 producer: group: rocketmq_group myrocketmq-config: my-topic: rocketmq_topic my-consumer-group: rocketmq_group_consumer
四:編寫消費者,消息生產者,消息實體類(自定義)
//Rocket消費者 @Slf4j @Component @RocketMQMessageListener(consumerGroup = "${myrocketmq-config.my-consumer-group}", topic = "${myrocketmq-config.my-topic}") public class RocketConsumer implements RocketMQListener<RocketmqVo> { @Override public void onMessage(RocketmqVo rocketmqVo) { log.info("收到RocketMQ消息:{}",rocketmqVo); } }
//Rocket生產者 @Slf4j @Component public class RocketProduce { @Value("${myrocketmq-config.my-topic}") private String mqTopic; @Value("${myrocketmq-config.my-consumer-group}") private String mqConsumerGroup; @Autowired private RocketMQTemplate mqTemplate; /** * 同步發送 * * @param title 發送消息 * @param data 消息內容 */ public void sync(String title, Object data) { String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date()); RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data); SendResult sendResult = mqTemplate.syncSend(mqTopic, rocketmqVo); log.info("同步發送:{}", rocketmqVo); log.info("同步發送消息結果:{}", sendResult); } /** * 異步發送 * * @param title 發送消息 * @param data 消息內容 */ public void async(String title, Object data) { String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date()); RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data); log.info("異步發送:{}", rocketmqVo); mqTemplate.asyncSend(mqTopic, rocketmqVo, new SendCallback() { @Override public void onSuccess(SendResult var1) { log.info("異步發送成功:{}", var1); } @Override public void onException(Throwable var1) { log.info("異步發送失敗:{}", var1.getMessage()); } }); } /** * 單向發送 * * @param title 發送消息 */ public void oneway(String title, Object data) { String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date()); RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data); mqTemplate.sendOneWay(mqTopic, rocketmqVo); log.info("單向發送:{}", rocketmqVo); } } //Rocket消息實體 @Data @AllArgsConstructor @NoArgsConstructor @ToString public class RocketmqVo { @Value("${myrocketmq-config.my-consumer-group}") private String group; @Value("${myrocketmq-config.my-topic}") private String topic; private String title; private String date; private Object data; public RocketmqVo(String title, String date, Object data) { this.title = title; this.date = date; this.data = data; } }
五:測試Controller
@RestController @RequestMapping("/mqtest") public class RocketController { @Resource private RocketProduce producer; @RequestMapping("/sendMessageSync/{msg}") public String sendMessageSync(@PathVariable("msg") String message) { producer.sync(message, null); return "消息發送完成"; } @RequestMapping("/sendMessageAsync/{msg}") public String sendMessageAsync(@PathVariable("msg") String message) { producer.async(message, null); return "消息發送完成"; } @RequestMapping("/sendMessageOneway/{msg}") public String sendMessageOneway(@PathVariable("msg") String message) { producer.oneway(message, null); return "消息發送完成"; } }
以上就是SpringBoot整合RocketMQ的方法詳解的詳細內容,更多關於SpringBoot整合RocketMQ的資料請關註WalkonNet其它相關文章!
推薦閱讀:
- 分佈式消息隊列RocketMQ概念詳解
- 一文徹底掌握RocketMQ 的存儲模型
- docker安裝RocketMQ的實現步驟
- 解決springboot集成rocketmq關於tag的坑
- RocketMQTemplate 註入失敗的解決