開發手冊

# 前言

本文檔為金蝶Apusic分佈式消息隊列for MQTT（Apusic Distributed Message Queue，簡稱：ADMQ for MQTT）的開發使用說明，幫助用戶快速學習如何使用金蝶Apusic分佈式消息隊列for MQTT進行開發。

# 適用對象

本文檔適用於IT信息化業務負責人、研發經理、軟件項目經理、軟件架構師、運維工程師。

# 相關文檔

瞭解更多ADMQ for MQTT產品相關的信息，請參閱以下ADMQ for MQTT產品手冊文檔集：

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

# 技術支持

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

  • 網址：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. 出現異常及錯誤的日誌、截圖等詳細信息

# 開發準備

# 連接參數

參數 說明 示例
Broker 地址 MQTT Broker 地址 localhost
端口 MQTT 端口 1883（TCP）、8883（SSL）、8083（WebSocket）
ClientID 客戶端唯一標識 device-001
username 用戶名 device-service
password 密碼 your-password

# 客戶端依賴

# Java

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>
1
2
3
4
5

# Python

pip install paho-mqtt
1

# JavaScript/Node.js

npm install mqtt
1

# Go

go get github.com/eclipse/paho.mqtt.golang
1

# Java 開發示例

# 生產者

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

public class MqttProducer {
    public static void main(String[] args) throws Exception {
        MqttClient client = new MqttClient("tcp://localhost:1883", "publisher-001");
        
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("device-service");
        options.setPassword("your-password".toCharArray());
        options.setCleanSession(true);
        
        client.connect(options);
        
        String payload = "{\"deviceId\":\"D001\",\"temperature\":24.5}";
        MqttMessage message = new MqttMessage(payload.getBytes("UTF-8"));
        message.setQos(1);
        message.setRetained(false);
        
        client.publish("sensor/D001/temperature", message);
        System.out.println("Message sent");
        
        client.disconnect();
        client.close();
    }
}
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

# 消費者

import org.eclipse.paho.client.mqttv3.*;

public class MqttConsumer {
    public static void main(String[] args) throws Exception {
        MqttClient client = new MqttClient("tcp://localhost:1883", "subscriber-001");
        
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("device-service");
        options.setPassword("your-password".toCharArray());
        options.setCleanSession(true);
        
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("Connection lost: " + cause.getMessage());
            }
            
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("Received: " + topic + " -> " + new String(message.getPayload(), "UTF-8"));
            }
            
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                // 用于 QoS 1/2 的发布确认
            }
        });
        
        client.connect(options);
        client.subscribe("sensor/+/temperature", 1);
        
        System.out.println("Subscribed");
    }
}
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

# 遺囑消息

MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("device-service");
options.setPassword("your-password".toCharArray());

// 设置遗嘱消息
options.setWill("device/D001/status", "offline".getBytes(), 1, true);

client.connect(options);

// 上线后发送在线状态
client.publish("device/D001/status", "online".getBytes(), 1, true);
1
2
3
4
5
6
7
8
9
10
11

# 會話保持

MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("device-service");
options.setPassword("your-password".toCharArray());
options.setCleanSession(false);  // 保持会话

client.connect(options);
client.subscribe("sensor/D001/temperature", 1);
1
2
3
4
5
6
7

# Python 開發示例

# 生產者

import paho.mqtt.client as mqtt

client = mqtt.Client(client_id="publisher-001")
client.username_pw_set("device-service", "your-password")

client.connect("localhost", 1883, 60)

payload = '{"deviceId":"D001","temperature":24.5}'
client.publish("sensor/D001/temperature", payload, qos=1, retain=False)
print("Message sent")

client.disconnect()
1
2
3
4
5
6
7
8
9
10
11
12

# 消費者

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    client.subscribe("sensor/+/temperature", qos=1)

def on_message(client, userdata, msg):
    print(f"Received: {msg.topic} -> {msg.payload.decode('utf-8')}")

client = mqtt.Client(client_id="subscriber-001")
client.username_pw_set("device-service", "your-password")
client.on_connect = on_connect
client.on_message = on_message

client.connect("localhost", 1883, 60)
client.loop_forever()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# JavaScript/Node.js 開發示例

# 生產者

const mqtt = require('mqtt');

const client = mqtt.connect('mqtt://localhost:1883', {
    clientId: 'publisher-001',
    username: 'device-service',
    password: 'your-password'
});

client.on('connect', () => {
    const payload = JSON.stringify({ deviceId: 'D001', temperature: 24.5 });
    client.publish('sensor/D001/temperature', payload, { qos: 1 }, (err) => {
        if (err) {
            console.error('Publish error:', err);
        } else {
            console.log('Message sent');
        }
        client.end();
    });
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 消費者

const mqtt = require('mqtt');

const client = mqtt.connect('mqtt://localhost:1883', {
    clientId: 'subscriber-001',
    username: 'device-service',
    password: 'your-password'
});

client.on('connect', () => {
    client.subscribe('sensor/+/temperature', { qos: 1 }, (err) => {
        if (err) {
            console.error('Subscribe error:', err);
        } else {
            console.log('Subscribed');
        }
    });
});

client.on('message', (topic, message) => {
    console.log(`Received: ${topic} -> ${message.toString()}`);
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# Go 開發示例

# 生產者

package main

import (
    "fmt"
    "time"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    opts := mqtt.NewClientOptions().
        AddBroker("tcp://localhost:1883").
        SetClientID("publisher-001").
        SetUsername("device-service").
        SetPassword("your-password")
    
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }
    
    payload := `{"deviceId":"D001","temperature":24.5}`
    token := client.Publish("sensor/D001/temperature", 1, false, payload)
    token.Wait()
    
    fmt.Println("Message sent")
    client.Disconnect(250)
}
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

# 消費者

package main

import (
    "fmt"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    opts := mqtt.NewClientOptions().
        AddBroker("tcp://localhost:1883").
        SetClientID("subscriber-001").
        SetUsername("device-service").
        SetPassword("your-password")
    
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }
    
    if token := client.Subscribe("sensor/+/temperature", 1, func(client mqtt.Client, msg mqtt.Message) {
        fmt.Printf("Received: %s -> %s\n", msg.Topic(), string(msg.Payload()))
    }); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }
    
    select {}
}
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

# 高級特性

# 保留消息

MqttMessage message = new MqttMessage(payload.getBytes());
message.setQos(1);
message.setRetained(true);  // 设置为保留消息
client.publish("sensor/D001/status", message);

// 清除保留消息
MqttMessage clearMessage = new MqttMessage(new byte[0]);
clearMessage.setRetained(true);
client.publish("sensor/D001/status", clearMessage);
1
2
3
4
5
6
7
8
9

# 共享訂閱

MQTT 5.0 共享訂閱格式：

$share/group1/sensor/+/temperature
1

同一共享組內的訂閱者輪流接收消息，實現負載均衡。

# WebSocket 接入

const client = mqtt.connect('ws://localhost:8083/mqtt', {
    clientId: 'web-client-001',
    username: 'device-service',
    password: 'your-password'
});
1
2
3
4
5

# SSL/TLS 接入

MqttClient client = new MqttClient("ssl://localhost:8883", "publisher-001");

MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("device-service");
options.setPassword("your-password".toCharArray());

// 配置 SSL 上下文
// options.setSocketFactory(sslSocketFactory);

client.connect(options);
1
2
3
4
5
6
7
8
9
10

# 最佳實踐

# ClientID 管理

  1. 全局唯一：每個客戶端的 ClientID 必須在集群內唯一
  2. 可讀性：建議使用有意義的前綴，如 device-、app-、service-
  3. 避免特殊字符：只使用字母、數字、下劃線、連字符

# QoS 選擇

場景 推薦 QoS
高頻 Telemetry 數據，可容忍丟失 QoS 0
一般業務消息 QoS 1
關鍵控制命令，要求恰好一次 QoS 2

# 會話管理

  1. Clean Session = true：適用於不需要離線消息的場景，資源佔用少
  2. Clean Session = false：適用於需要保持訂閱和接收離線消息的場景
  3. MQTT 5.0：使用 Clean Start 和 Session Expiry Interval 精確控制會話行為

# 異常處理

  1. 斷線重連：實現連接丟失後的自動重連機制
  2. 消息去重：QoS 1 消息可能重複，業務側需實現冪等處理
  3. 超時處理：合理設置連接超時和 Keepalive 間隔

# 性能優化

  1. 批量發送：減少頻繁的小消息發送
  2. 合理設置 Keepalive：避免過短的心跳間隔增加網絡負載
  3. 使用共享訂閱：多個消費者均衡處理消息
  4. 限制訂閱數量：單個客戶端訂閱過多主題會影響性能
  5. 避免通配符濫用：儘量使用精確主題匹配
编辑页面 (opens new window)
#開發手冊

← 遷移手冊→