快速入門

# 前言

本文檔為金蝶Apusic分散式訊息佇列for RocketMQ（Apusic Distributed Message Queue for RocketMQ，簡稱：ADMQ for RocketMQ）的快速使用手冊，提供產品介紹，安裝部署產品的基本步驟及驗證方式。

# 適用對象

本文檔適用於IT資訊化業務負責人、研發經理、軟體專案經理、軟體架構師、運維工程師。

# 相關文檔

了解更多ADMQ for RocketMQ產品相關的資訊，請參閱以下ADMQ for RocketMQ產品手冊文件集：

序號 手冊文檔 說明
1 金蝶Apusic分散式訊息佇列for RocketMQ 快速使用手冊 簡單介紹了如何快速上手使用ADMQ for RocketMQ 。
2 金蝶Apusic分散式訊息佇列for RocketMQ 安裝手冊 詳細介紹如何在各作業系統上安裝ADMQ for RocketMQ，以及ADMQ for RocketMQ服務啟停等操作。
3 金蝶Apusic分散式訊息佇列for RocketMQ 訊息引擎使用者手冊 詳細介紹 ADMQ for RocketMQ 訊息引擎相關功能的使用、配置、管理及配套工具的使用方法。
4 金蝶Apusic分散式訊息佇列for RocketMQ 管控台使用者手冊 詳細介紹ADMQ for RocketMQ管控台相關功能的使用和操作說明。
5 金蝶Apusic分散式訊息佇列for RocketMQ 開發手冊 詳細介紹基於各開發語言進行ADMQ for RocketMQ客戶端應用開發的說明。
6 金蝶Apusic分散式訊息佇列for RocketMQ 遷移手冊 詳細介紹從RocketMQ遷移到ADMQ for RocketMQ的說明。
7 金蝶Apusic分散式訊息佇列for RocketMQ 運維手冊 詳細介紹ADMQ for RocketMQ的監控、運維、安全加固等運維說明。
8 金蝶Apusic分散式訊息佇列for RocketMQ 效能最佳化手冊 詳細介紹ADMQ for RocketMQ效能調優的說明。

# 技術支援

ADMQ for RocketMQ產品提供全面的技術支援服務，您可以透過以下方式獲得技術支援：

  • 網址：www.apusic.com
  • 電話：400-855-5800
  • 郵箱：support@apusic.com
  • 金蝶雲社群：https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

您在取得技術支援時，請提供如下資訊：

  1. 您的姓名

  2. 公司資訊與聯絡方式

  3. 作業系統及其版本

  4. 產品版本號

  5. 出現異常及錯誤的日誌、截圖等詳細資訊

# 快速部署

# 環境準備

元件 最低要求
CPU 4 核
記憶體 16 GB
磁碟 200 GB SSD
作業系統 CentOS 7+ / Ubuntu 18.04+

# 一鍵部署

mkdir -p /apusic
cd /apusic
# 1.解壓安裝包
#tar zxvf ADMQ-V2.0.6-RocketMQ-構建日期.tar.gz -C /apusic/
tar zxvf ADMQ-V2.0.6.534-RocketMQ-20260624.tar.gz -C /apusic/
cd /apusic
# 建議nameserver broker元件資料夾結構分開
unzip rocketmq-V5.3.4-all.zip
mv rocketmq-V5.3.4-all nameserver
cd /apusic/nameserver

unzip rocketmq-V5.3.4-all.zip
mv rocketmq-V5.3.4-all broker
cd /apusic/broker

# 2. 啟動 NameServer
bin/rocketmq start nameserver

# 3. 啟動 Broker
bin/rocketmq start broker

# 4. 驗證狀態
bin/mqadmin clusterList -n localhost:9876
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 管控台部署

# 1. 解壓管控台
tar zxvf admq-manager-V2.0.6.tar.gz -C /apusic/
cd /apusic/admq-manager-V2.0.6

# 2. 啟動管控台
bin/admq-service start
1
2
3
4
5
6

# 快速驗證

# 訪問管控台

地址：http://<管控台IP>:12305;https://<管控台IP>:12306
預設帳號：admq / 11111111
1
2

# 建立測試資源

  1. 建立主題

    • 進入「資源管理」-「主題管理」頁面
    • 點選「新建」，輸入主題名稱（如：test-topic）
    • 配置讀寫佇列數（如：3）
    • 確認建立

  2. 建立訂閱組

    • 進入「資源管理」-「訂閱組管理」頁面
    • 點選「新建」，輸入訂閱組名稱（如：group）
    • 配置訂閱組資訊
    • 確認建立

  3. 檢視訊息傳送

    • 進入「訊息查詢」頁面

    • 選擇主題：test-topic

    • 可檢視該 Topic 的訊息

# 傳送和接收訊息

# 使用命令列工具

# 傳送訊息
bin/mqadmin sendMessage -n localhost:9876 -t test-topic -p "Hello ADMQ for RocketMQ!"

# 檢視訊息
bin/mqadmin queryMsgByKey -n localhost:9876 -t test-topic -k test-key
1
2
3
4
5

# 使用管控台傳送訊息

  1. 進入「資源管理」-「Topic 管理」頁面
  2. 選擇 test-topic，點選「傳送訊息」
  3. 填寫訊息內容、Tag、Key 等資訊
  4. 點選傳送

# 使用開源SDK實現訊息收發（Java SDK為例）

# 概覽

目前RocketMQ相容開源特性，根據底層通訊協議的差異主要支援兩個系列的客戶端SDK，分別是Remoting協定和gRPC協定 關於 Remoting 協定 SDK 和 gRPC 協定 SDK 的對比參考如下：

對比項 Remoting 協定SDK gRPC 協定SDK
多語言支援 Java為主，其他語言為第三方倉庫實現 Java/C/C++/.NET/Go/Rust
介面範圍 Producer
PushConsumer
PullConsumer
LitePullConsumer
Admin		 Producer
PushConsumer(僅Java)
SimpleConsumer
相容版本 相容4.x、5.x版本服務端 僅支援5.x 版本服務端
# 生產訊息

建立訊息生產者

// 實例化訊息生產者Producer
DefaultMQProducer producer = new DefaultMQProducer(
  	 namespace, 
  	 groupName,
  	 new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL權限
);
// 設定NameServer的地址
producer.setNamesrvAddr(nameserver);
// 啟動Producer實例
producer.start();
1
2
3
4
5
6
7
8
9
10

傳送訊息 傳送訊息由多種方式，同步傳送、非同步傳送、單向傳送等。

  • 同步訊息
 for (int i = 0; i < 10; i++) {
 // 建立訊息實例，設定topic和訊息內容
 Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
 // 傳送訊息
 SendResult sendResult = producer.send(msg);
 System.out.printf("%s%n", sendResult);
 }
1
2
3
4
5
6
7
  • 非同步訊息
// 設定傳送失敗後不重試
producer.setRetryTimesWhenSendAsyncFailed(0);
// 設定傳送訊息的數量
int messageCount = 10;
final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
for (int i = 0; i < messageCount; i++) {
try {
		 final int index = i;
		 // 建立訊息實體，設定topic和訊息內容
		 Message msg = new Message(topic_name, "TAG", ("Hello rocketMq " + index).getBytes(RemotingHelper.DEFAULT_CHARSET));
		 producer.send(msg, new SendCallback() {
				 @Override
				 public void onSuccess(SendResult sendResult) {
						 // 訊息傳送成功邏輯
						 countDownLatch.countDown();
						 System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
				 }

				 @Override
				 public void onException(Throwable e) {
						 // 訊息傳送失敗邏輯
						 countDownLatch.countDown();
						 System.out.printf("%-10d Exception %s %n", index, e);
						 e.printStackTrace();
				 }
		 });
 } catch (Exception e) {
		 e.printStackTrace();
 }
}
countDownLatch.await(5, TimeUnit.SECONDS);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
  • 單向傳送
for (int i = 0; i < 10; i++) {
 // 建立訊息實例，設定topic和訊息內容
 Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
 // 傳送單向訊息
 producer.sendOneway(msg);
}
1
2
3
4
5
6
# 消費訊息（Remoting協定為例）

建立消費者 支援 push 和 pull 兩種消費模式

  • push消費者
 // 實例化消費者
DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
 namespace,                                                  
 groupName,                                              
 new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL權限
// 設定NameServer的地址
pushConsumer.setNamesrvAddr(nameserver);
1
2
3
4
5
6
7
  • pull消費者
// 實例化消費者
DefaultLitePullConsumer pullConsumer = new DefaultLitePullConsumer(
namespace,                                               
 groupName,                                             
 new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
// 設定NameServer的地址
pullConsumer.setNamesrvAddr(nameserver);
// 設定從第一個偏移量開始消費
pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
1
2
3
4
5
6
7
8
9

訂閱訊息 根據消費模式不同，訂閱方式也有所區別。

  • push訂閱
// 訂閱topic
pushConsumer.subscribe(topic_name, "*");
// 註冊回撥實現類來處理從broker拉取回來的訊息
pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
 // 訊息處理邏輯
 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
 // 標記該訊息已經被成功消費, 根據消費情況，返回處理狀態
 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 啟動消費者實例
pushConsumer.start();
1
2
3
4
5
6
7
8
9
10
11
  • pull訂閱
 // 訂閱topic
pullConsumer.subscribe(topic_name, "*");
// 啟動消費者實例
pullConsumer.start();
try {
 System.out.printf("Consumer Started.%n");
 while (true) {
		 // 拉取訊息
		 List<MessageExt> messageExts = pullConsumer.poll();
		 System.out.printf("%s%n", messageExts);
 }
} finally {
  	 pullConsumer.shutdown();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 客戶端快速接入

# Java 客戶端示例

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class QuickStart {
    public static void main(String[] args) throws Exception {
        // 生產者
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        
        // 傳送訊息
        Message msg = new Message("test-topic", "tag-a", "test-key", 
            "Hello ADMQ for RocketMQ!".getBytes("UTF-8"));
        producer.send(msg);
        System.out.println("Message sent");
        producer.shutdown();
        
        // 消費者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("test-topic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                    ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("Received: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer started");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# Go 客戶端示例

package main

import (
    "context"
    "fmt"
    "github.com/apache/rocketmq-clients/golang/v2"
)

func main() {
    // 生產者
    producer, err := golang.NewProducer(
        golang.WithEndpoints("localhost:9876"),
        golang.WithConsumerGroup("test-group"),
    )
    if err != nil {
        panic(err)
    }
    
    err = producer.Start()
    if err != nil {
        panic(err)
    }
    defer producer.Shutdown()
    
    msg := golang.NewMessage("test-topic", []byte("Hello ADMQ for RocketMQ!"))
    resp, err := producer.Send(context.TODO(), msg)
    if err != nil {
        panic(err)
    }
    fmt.Printf("Message sent: %s\n", resp[0].MsgId)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
编辑页面 (opens new window)
#快速使用手冊

← 管控台使用者手冊→