工作了這么多年，rocketmq還沒(méi)有用過(guò)，由于現(xiàn)在的工作中涉及到了，周六吃完午飯就開(kāi)始搞，結(jié)果到現(xiàn)在3點(diǎn)鐘才把環(huán)境弄好，測(cè)試代碼搞起。
整個(gè)流程分成兩步
- 安裝簡(jiǎn)單的rocket環(huán)境
- 起springboot項(xiàng)目測(cè)試
參考文章：
https://blog.csdn.net/baidu_33256174/article/details/129599300
簡(jiǎn)單的rocket環(huán)境
1. 制作rocketmq:4.5.0的鏡像
- 從https://github.com/apache/rocketmq-docker拉取最新的代碼，解壓
- cd image-build
- ./build-image.sh 4.5.0 centos 注意alpine在M1上失敗
? ~/ docker images
REPOSITORY TAG IMAGE ID CREATED SIZE
apache/rocketmq 4.5.0 04b1d4c1d001 About an hour ago 1.28GB
bitnami/mysql 5.7 4164f4e78f8e 2 days ago 415MB
redis latest 37a7207b0faf 3 days ago 149MB
apacherocketmq/rocketmq-dashboard latest eae6c5db5d11 20 months ago 738MB
candice0630/rocketmq-console-ng 2.0 c3494a6e4d86 2 years ago 357MB
apacherocketmq/rocketmq 4.5.0 fa3c9d27c922 3 years ago 493MB
? ~/
2. 制作rocketmq-console-ng鏡像
參考 https://www.jianshu.com/p/6ad529a16677
偷懶直接 docker pull candice0630/rocketmq-console-ng:2.0
文章提供的步驟如下
3. 起環(huán)境
1. 新建一個(gè)rocket的目錄，創(chuàng)建broker.conf 文件
# 所屬集群名稱，如果節(jié)點(diǎn)較多可以配置多個(gè)
brokerClusterName = DefaultCluster
#broker名稱，master和slave使用相同的名稱，表明他們的主從關(guān)系
brokerName = broker-a
#0表示Master，大于0表示不同的slave
brokerId = 0
#表示幾點(diǎn)做消息刪除動(dòng)作，默認(rèn)是凌晨4點(diǎn)
deleteWhen = 04
#在磁盤(pán)上保留消息的時(shí)長(zhǎng)，單位是小時(shí)
fileReservedTime = 48
#有三個(gè)值：SYNC_MASTER，ASYNC_MASTER，SLAVE；同步和異步表示Master和Slave之間同步數(shù)據(jù)的機(jī)制；
brokerRole = ASYNC_MASTER
#刷盤(pán)策略，取值為：ASYNC_FLUSH，SYNC_FLUSH表示同步刷盤(pán)和異步刷盤(pán)；SYNC_FLUSH消息寫(xiě)入磁盤(pán)后才返回成功
狀態(tài)，ASYNC_FLUSH不需要；
flushDiskType = ASYNC_FLUSH
# 設(shè)置broker節(jié)點(diǎn)所在服務(wù)器的ip地址、物理ip，不能用127.0.0.1、localhost、docker內(nèi)網(wǎng)ip
brokerIP1 = 192.168.18.102
2. 創(chuàng)建rocketmq.yml文件
version: '3'
services:
namesrv:
image: apache/rocketmq:4.5.0
container_name: rmqnamesrv
ports:
- 9876:9876
command: sh mqnamesrv
broker:
image: apache/rocketmq:4.5.0
container_name: rmqbroker
ports:
- 10909:10909
- 10911:10911
- 10912:10912
volumes:
- /Users/lixi/rocketmq/broker.conf:/home/rocketmq/rocketmq-4.5.0/conf/broker.conf
command: sh mqbroker -n namesrv:9876 -c /home/rocketmq/rocketmq-4.5.0/conf/broker.conf
depends_on:
- namesrv
mqconsole:
image: candice0630/rocketmq-console-ng:2.0
container_name: rmqdashboard
ports:
- 8181:8080
environment:
JAVA_OPTS: -Drocketmq.config.namesrvAddr=namesrv:9876 -Drocketmq.config.isVIPChannel=false
depends_on:
- namesrv
使用docker-compose -f rocketmq.yml up -d啟動(dòng)后，正常會(huì)有下面的容器
3. 查看 http://localhost:8181
文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-553356.html
2. springboot項(xiàng)目測(cè)試
0. pom
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.4.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>lrocket</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>lrocket</name>
<description>lrocket</description>
<properties>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
1. 創(chuàng)建配置文件
# rocketmq 配置項(xiàng)，對(duì)應(yīng) RocketMQProperties 配置類
rocketmq:
name-server: 127.0.0.1:9876 # RocketMQ Namesrv
# Producer 配置項(xiàng)
producer:
group: demo-producer-group # 生產(chǎn)者分組
send-message-timeout: 3000 # 發(fā)送消息超時(shí)時(shí)間，單位：毫秒。默認(rèn)為 3000 。
compress-message-body-threshold: 4096 # 消息壓縮閥值，當(dāng)消息體的大小超過(guò)該閥值后，進(jìn)行消息壓縮。默認(rèn)為 4 * 1024B
max-message-size: 4194304 # 消息體的最大允許大小。。默認(rèn)為 4 * 1024 * 1024B
retry-times-when-send-failed: 2 # 同步發(fā)送消息時(shí)，失敗重試次數(shù)。默認(rèn)為 2 次。
retry-times-when-send-async-failed: 2 # 異步發(fā)送消息時(shí)，失敗重試次數(shù)。默認(rèn)為 2 次。
retry-next-server: false # 發(fā)送消息給 Broker 時(shí)，如果發(fā)送失敗，是否重試另外一臺(tái) Broker 。默認(rèn)為 false
access-key: # Access Key ，可閱讀 https://github.com/apache/rocketmq/blob/master/docs/cn/acl/user_guide.md 文檔
secret-key: # Secret Key
enable-msg-trace: true # 是否開(kāi)啟消息軌跡功能。默認(rèn)為 true 開(kāi)啟?？砷喿x https://github.com/apache/rocketmq/blob/master/docs/cn/msg_trace/user_guide.md 文檔
customized-trace-topic: RMQ_SYS_TRACE_TOPIC # 自定義消息軌跡的 Topic 。默認(rèn)為 RMQ_SYS_TRACE_TOPIC 。
# Consumer 配置項(xiàng)
consumer:
listeners: # 配置某個(gè)消費(fèi)分組，是否監(jiān)聽(tīng)指定 Topic 。結(jié)構(gòu)為 Map<消費(fèi)者分組, <Topic, Boolean>> 。默認(rèn)情況下，不配置表示監(jiān)聽(tīng)。
test-consumer-group:
topic1: false # 關(guān)閉 test-consumer-group 對(duì) topic1 的監(jiān)聽(tīng)消費(fèi)
2. 寫(xiě)生產(chǎn)者
package com.example.lrocket.controler;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.time.LocalDateTime;
@RestController
public class RocketController {
@Resource
private RocketMQTemplate rocketMQTemplate;
// 延時(shí)消息，RocketMQ支持這幾個(gè)級(jí)別的延時(shí)消息，自定義需要修改broker配置文件
// 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
@GetMapping("/rocket/delayMsg/send")
public String rocketDelayMsgSend() {
LocalDateTime currentDateTime = LocalDateTime.now();
rocketMQTemplate.syncSend("rocket-topic-2:tag-2", MessageBuilder.withPayload(currentDateTime.toString()).build(), 2000, 3);
return currentDateTime.toString();
}
@GetMapping("/rocket/send")
public String rocketMsgSend() {
LocalDateTime currentDateTime = LocalDateTime.now();
rocketMQTemplate.syncSend("rocket-topic-2:tag-2", MessageBuilder.withPayload(currentDateTime.toString()).build(), 2000, 0);
return currentDateTime.toString();
}
}
3. 寫(xiě)消費(fèi)者
package com.example.lrocket.listener;
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.Component;
import org.springframework.stereotype.Service;
@Component
public class RokcetServiceListener {
@Service
@RocketMQMessageListener(consumerGroup = "consumer-group-1", topic = "rocket-topic-2")
public class Consumer1 implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("consumer1 rocket收到消息: " + s);
}
}
//MessageModel.BROADCASTING 廣播消息模式
@Service
@RocketMQMessageListener(consumerGroup = "consumer-group-2", topic = "rocket-topic-2", selectorExpression = "tag-2", messageModel = MessageModel.BROADCASTING)
public class Consumer2 implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
System.out.println("consumer2 rocket收到消息:" + s);
}
}
}
4. 測(cè)試
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-553356.html
到了這里，關(guān)于MAC M1上docker rocketmq簡(jiǎn)單環(huán)境搭建和代碼的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
