开发手册

# 前言

本文档为金蝶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)
#开发手册

← 迁移手册→