Development Manual

# Preface

This document is the development guide for the Kingdee Apusic Distributed Message Queue for MQTT (Apusic Distributed Message Queue, ADMQ for MQTT), helping users quickly learn how to develop with ADMQ for MQTT.

# Target Audience

This document is intended for IT business leaders, R&D managers, software project managers, software architects, and operations engineers.

To learn more about ADMQ for MQTT, please refer to the following ADMQ for MQTT product documentation set:

No. Document Description
1 ADMQ for MQTT Quick Start Guide Briefly introduces how to quickly get started with ADMQ for MQTT.
2 ADMQ for MQTT Installation Manual Details how to install ADMQ for MQTT on various operating systems and how to start/stop ADMQ for MQTT services.
3 ADMQ for MQTT Message Engine User Manual Details the usage, configuration, management, and supporting tools of ADMQ for MQTT message engine features.
4 ADMQ for MQTT Console User Manual Details the usage and operation instructions of the ADMQ for MQTT console.
5 ADMQ for MQTT Development Manual Details ADMQ for MQTT client application development based on various programming languages.
6 ADMQ for MQTT Migration Manual Details the migration from MQTT Broker to ADMQ for MQTT.
7 ADMQ for MQTT Operations Manual Details ADMQ for MQTT monitoring, operations, security hardening, and other operational tasks.
8 ADMQ for MQTT Performance Tuning Manual Details ADMQ for MQTT performance tuning guidelines.

# Technical Support

ADMQ for MQTT provides comprehensive technical support services. You can obtain technical support through the following channels:

  • Website: www.apusic.com
  • Phone: 400-855-5800
  • Email: support@apusic.com
  • Kingdee Cloud Community: https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

When contacting technical support, please provide the following information:

  1. Your name

  2. Company information and contact details

  3. Operating system and its version

  4. Product version number

  5. Detailed information such as logs and screenshots of the exception or error

# Development Preparation

# Connection Parameters

Parameter Description Example
Broker Address MQTT Broker address localhost
Port MQTT port 1883 (TCP), 8883 (SSL), 8083 (WebSocket)
ClientID Unique client identifier device-001
username Username device-service
password Password your-password

# Client Dependencies

# 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 Development Example

# Producer

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

# Consumer

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) {
                // Used for QoS 1/2 publish acknowledgment
            }
        });
        
        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

# Will Message

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

// Set will message
options.setWill("device/D001/status", "offline".getBytes(), 1, true);

client.connect(options);

// Send online status after going online
client.publish("device/D001/status", "online".getBytes(), 1, true);
1
2
3
4
5
6
7
8
9
10
11

# Session Persistence

MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("device-service");
options.setPassword("your-password".toCharArray());
options.setCleanSession(false);  // Keep session

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

# Python Development Example

# Producer

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

# Consumer

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 Development Example

# Producer

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

# Consumer

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 Development Example

# Producer

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

# Consumer

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

# Advanced Features

# Retained Messages

MqttMessage message = new MqttMessage(payload.getBytes());
message.setQos(1);
message.setRetained(true);  // Set as retained message
client.publish("sensor/D001/status", message);

// Clear retained 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

# Shared Subscription

MQTT 5.0 shared subscription format:

$share/group1/sensor/+/temperature
1

Subscribers within the same shared group receive messages in a round-robin fashion, achieving load balancing.

# WebSocket Connection

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 Connection

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

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

// Configure SSL context
// options.setSocketFactory(sslSocketFactory);

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

# Best Practices

# ClientID Management

  1. Globally unique: Each client's ClientID must be unique within the cluster
  2. Readability: Use meaningful prefixes such as device-, app-, service-
  3. Avoid special characters: Use only letters, numbers, underscores, and hyphens

# QoS Selection

Scenario Recommended QoS
High-frequency Telemetry data, loss-tolerant QoS 0
General business messages QoS 1
Critical control commands requiring exactly-once delivery QoS 2

# Session Management

  1. Clean Session = true: Suitable for scenarios that do not require offline messages, lower resource consumption
  2. Clean Session = false: Suitable for scenarios that need to maintain subscriptions and receive offline messages
  3. MQTT 5.0: Use Clean Start and Session Expiry Interval to precisely control session behavior

# Exception Handling

  1. Reconnection: Implement automatic reconnection mechanism after connection loss
  2. Message deduplication: QoS 1 messages may be duplicated, implement idempotent processing on the business side
  3. Timeout handling: Set connection timeout and Keepalive interval appropriately

# Performance Optimization

  1. Batch sending: Reduce frequent small message sends
  2. Set Keepalive appropriately: Avoid excessively short heartbeat intervals that increase network load
  3. Use shared subscriptions: Multiple consumers balance message processing
  4. Limit subscription count: A single client subscribing to too many topics affects performance
  5. Avoid wildcard abuse: Prefer exact topic matching where possible
编辑页面 (opens new window)
#Development Manual

← Migration Manual→