RocketMQ(4.7.0)單機與集群搭建,控制台搭建,並與springboot整合

參考書籍: RocketMQ實戰與原理解析: 作者楊開元

一 單機部署

親測jdk11有問題,無法啟動,最好用jdk8

修改shell腳本:

修改合適的內存:

先啟動NameServer再啟動一個Broker:

日志默認位置: ~/logs/rocketmqlogs

測試:

發送demo: sh tools.sh org.apache.rocketmq.example.quickstart.Producer
接受demo: sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
注意:每個窗口都必須先設置NAMESRV_ADDR,否則無法連接: export NAMESRV_ADDR=localhost:9876

關閉

sh mqshutdown namesrv
sh mqshutdown broker

二 集群部署

先部署兩台服務器到兩個服務器上

結構:

服務器a,服務器b, 兩個NameServer, 兩個brokera,b,兩台服務器互為主從,即服務器a上面有brokera和brokerb的salve

修改配置文件

位置: rocketmq-all-4.7.0/conf/2m-2s-async

一台修改broker-a.properties broker-b-s.properties,即brokera的master和brokerb的salve,另一台反之:

# NemeServer地址,可以有多個
namesrvAddr=192.168.159.128:9876;192.168.159.129:9876
# 集群名稱
brokerClusterName=DefaultCluster
# broker名稱,主從的名稱相同表明相互關聯,為同一個broker
brokerName=broker-b
# broker的id,一個master可以有多個slave,id為0表示主,大於0表示不同的從
brokerId=0
# 在幾點做刪除消息動作,比如04: 凌晨四點
deleteWhen=04
# 在磁盤保存消息的時長,超過會自動刪除,在上面配置刪除的時間點刪除
fileReservedTime=48
# 角色有三種:
#ASYNC_MASTER, 消息同步開始就返回成功狀態
#SYNC_MASTER  表示消息同步完成后再發送完成狀態
#SLAVE   從
brokerRole=ASYNC_MASTER
# 刷盤策略,同步和異步
flushDiskType=ASYNC_FLUSH
# 監聽端口,一台機器上多個Broker要確保端口不沖突
listenPort=10921
# 存儲消息以及一些配置信息的根目錄
storePathRootDir=/home/rocketmq/store-b

啟動

先啟動兩個NameServer
再啟動四個Broker,后面用 -c 跟上對應的配置文件:
nohup sh bin/mqbroker -c conf/2m-2s-async/broker-b-s.properties &
注意: 端口一定要提前放開

三 控制台搭建

下載解壓

wget https://github.com/apache/rocketmq-externals/archive/master.zip
解壓 unzip

編譯打包

之后進入目錄: cd rocketmq-externals-master/rocketmq-console
使用maven編譯打包:
mvn clean package
注意: 如果報錯就刪除測試代碼:
rm src/test -rf #刪除src 目錄下的 test
如果慢就編輯maven的setting,使用阿里雲的鏡像地址

運行

編譯后找到target下面的jar包,其實是一個springboot項目,啟動
nohup java -jar rocketmq-console-ng-1.0.0.jar --server.port=8561 --rocketmq.config.namesrvAddr=a:9876;b:9876 &
nohup 命令 & :后台啟動,防止session或者退出,打斷程序運行
成功運行:

四 springboot整合

添加start依賴

=implementation group: 'org.apache.rocketmq', name: 'rocketmq-spring-boot-starter', version: '2.0.4'

yml配置

rocketmq:
  name-server: 192.168.159.128:9876;192.168.159.129:9876
  producer:
    group: houzheng
    send-message-timeout: 30000

producer

    // springboot會自動注入,可直接使用
    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @GetMapping("send/{id}")
    public String send(@PathVariable("id") String id){
        rocketMQTemplate.send("test-topic-01",MessageBuilder.withPayload(new User(id,"侯征")).build());
        return "SUCESS";
    }

consumer

@Component
@RocketMQMessageListener(topic = "test-topic-01", consumerGroup = "my-consumer_test-topic-01")
public class UserConsumer implements RocketMQListener<User> {
    @Override
    public void onMessage(User message) {
        System.out.println("接受到消息:"+message.toString());
    }
}

測試

五 docker快速部署