Development Manual

# Preface

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

# Target Audience

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

For more information about ADMQ for RabbitMQ products, please refer to the following ADMQ for RabbitMQ product documentation set:

No.	Document	Description
1	Kingdee Apusic Distributed Message Queue for RabbitMQ Quick Start Guide	Briefly introduces how to quickly get started with ADMQ for RabbitMQ.
2	Kingdee Apusic Distributed Message Queue for RabbitMQ Installation Guide	Detailed instructions on installing ADMQ for RabbitMQ on various operating systems, as well as ADMQ for RabbitMQ service start/stop operations.
3	Kingdee Apusic Distributed Message Queue for RabbitMQ Message Engine User Guide	Detailed description of the usage, configuration, management, and supporting tools of the ADMQ for RabbitMQ message engine.
4	Kingdee Apusic Distributed Message Queue for RabbitMQ Console User Guide	Detailed description of the usage and operation of the ADMQ for RabbitMQ management console.
5	Kingdee Apusic Distributed Message Queue for RabbitMQ Development Guide	Detailed instructions on developing ADMQ for RabbitMQ client applications using various programming languages.
6	Kingdee Apusic Distributed Message Queue for RabbitMQ Migration Guide	Detailed instructions on migrating from RabbitMQ to ADMQ for RabbitMQ.
7	Kingdee Apusic Distributed Message Queue for RabbitMQ Operations Guide	Detailed description of ADMQ for RabbitMQ monitoring, operations, security hardening, and other maintenance tasks.
8	Kingdee Apusic Distributed Message Queue for RabbitMQ Performance Optimization Guide	Detailed description of ADMQ for RabbitMQ performance tuning.

# Technical Support

ADMQ for RabbitMQ 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 seeking technical support, please provide the following information:

Your name
Company information and contact details
Operating system and version
Product version number
Detailed information such as logs and screenshots of exceptions and errors

# Development Preparation

# Connection Parameters

Parameter	Description	Example
host	Server address	localhost
port	Port	5672 (AMQP) / 5671 (AMQP over SSL)
username	Username	guest
password	Password	guest
virtualHost	Virtual host	/

# Client Dependencies

# Java

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.21.0</version>
</dependency>

1
2
3
4
5

# Python

pip install pika

# Go

go get github.com/rabbitmq/amqp091-go

# Node.js

npm install amqplib

# Java Development Example

# Producer

import com.rabbitmq.client.*;

public class Producer {
    private static final String EXCHANGE_NAME = "demo.exchange";
    private static final String ROUTING_KEY = "demo.key";
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
            
            // 发送消息
            String message = "Hello ADMQ for RabbitMQ!";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes("UTF-8"));
            
            System.out.println("Sent: " + message);
        }
    }
}

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

# Consumer

import com.rabbitmq.client.*;

public class Consumer {
    private static final String QUEUE_NAME = "demo.queue";
    private static final String EXCHANGE_NAME = "demo.exchange";
    private static final String ROUTING_KEY = "demo.key";
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
        
        // 创建消费者
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received: " + message);
            
            // 手动确认
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        
        // 消费消息（关闭自动确认）
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

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

# Transactional Messages

channel.txSelect();  // 开启事务

try {
    channel.basicPublish(EXCHANGE, ROUTING_KEY, null, message1.getBytes());
    channel.basicPublish(EXCHANGE, ROUTING_KEY, null, message2.getBytes());
    channel.txCommit();  // 提交事务
} catch (Exception e) {
    channel.txRollback();  // 回滚事务
}

1
2
3
4
5
6
7
8
9

# Confirm Mode

channel.confirmSelect();  // 开启 Confirm 模式

channel.basicPublish(EXCHANGE, ROUTING_KEY, null, message.getBytes());

// 同步等待确认
if (channel.waitForConfirms()) {
    System.out.println("Message confirmed");
}

// 异步确认
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        System.out.println("Message ack: " + deliveryTag);
    }
    
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        System.out.println("Message nack: " + deliveryTag);
    }
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# Python Development Example

# Producer

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost', port=5672,
                              credentials=pika.PlainCredentials('guest', 'guest')))
channel = connection.channel()

channel.exchange_declare(exchange='demo.exchange', exchange_type='direct', durable=True)

channel.basic_publish(exchange='demo.exchange',
                      routing_key='demo.key',
                      body='Hello ADMQ for RabbitMQ!',
                      properties=pika.BasicProperties(delivery_mode=2))

print("Sent message")
connection.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# Consumer

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost', port=5672,
                              credentials=pika.PlainCredentials('guest', 'guest')))
channel = connection.channel()

channel.queue_declare(queue='demo.queue', durable=True)
channel.queue_bind(queue='demo.queue', exchange='demo.exchange', routing_key='demo.key')

def callback(ch, method, properties, body):
    print(f"Received {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='demo.queue', on_message_callback=callback, auto_ack=False)
print('Waiting for messages...')
channel.start_consuming()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# Go Development Example

# Producer

package main

import (
    "fmt"
    "github.com/rabbitmq/amqp091-go"
    "log"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()
    
    ch, err := conn.Channel()
    if err != nil {
        log.Fatal(err)
    }
    defer ch.Close()
    
    err = ch.ExchangeDeclare("demo.exchange", "direct", true, false, false, false, nil)
    if err != nil {
        log.Fatal(err)
    }
    
    body := "Hello ADMQ for RabbitMQ!"
    err = ch.Publish("demo.exchange", "demo.key", false, false,
        amqp091.Publishing{
            DeliveryMode: amqp091.Persistent,
            ContentType:  "text/plain",
            Body:         []byte(body),
        })
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Println("Sent:", body)
}

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

# Consumer

package main

import (
    "fmt"
    "github.com/rabbitmq/amqp091-go"
    "log"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()
    
    ch, err := conn.Channel()
    if err != nil {
        log.Fatal(err)
    }
    defer ch.Close()
    
    q, err := ch.QueueDeclare("demo.queue", true, false, false, false, nil)
    if err != nil {
        log.Fatal(err)
    }
    
    err = ch.QueueBind(q.Name, "demo.key", "demo.exchange", false, nil)
    if err != nil {
        log.Fatal(err)
    }
    
    msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Println("Waiting for messages...")
    for msg := range msgs {
        fmt.Printf("Received: %s\n", msg.Body)
        msg.Ack(false)
    }
}

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
42

# Node.js Development Example

# Producer

const amqp = require('amqplib');

async function produce() {
    const conn = await amqp.connect('amqp://guest:guest@localhost:5672');
    const ch = await conn.createChannel();
    
    await ch.assertExchange('demo.exchange', 'direct', { durable: true });
    
    const msg = 'Hello ADMQ for RabbitMQ!';
    ch.publish('demo.exchange', 'demo.key', Buffer.from(msg), { persistent: true });
    
    console.log('Sent:', msg);
    
    await ch.close();
    await conn.close();
}

produce().catch(console.error);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# Consumer

const amqp = require('amqplib');

async function consume() {
    const conn = await amqp.connect('amqp://guest:guest@localhost:5672');
    const ch = await conn.createChannel();
    
    await ch.assertQueue('demo.queue', { durable: true });
    await ch.bindQueue('demo.queue', 'demo.exchange', 'demo.key');
    
    console.log('Waiting for messages...');
    
    ch.consume('demo.queue', (msg) => {
        console.log('Received:', msg.content.toString());
        ch.ack(msg);
    }, { noAck: false });
}

consume().catch(console.error);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# Advanced Features

# Delayed Messages

Use dead letter exchange + TTL to implement delayed queues:

// 创建死信交换机和队列
channel.exchangeDeclare("dlx.exchange", "direct", true);
channel.queueDeclare("dlx.queue", true, false, false, null);
channel.queueBind("dlx.queue", "dlx.exchange", "dlx.key");

// 创建延迟队列（设置 TTL 和死信参数）
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000);  // 60秒延迟
args.put("x-dead-letter-exchange", "dlx.exchange");
args.put("x-dead-letter-routing-key", "dlx.key");

channel.queueDeclare("delay.queue", true, false, false, args);

// 发送消息到延迟队列
channel.basicPublish("", "delay.queue", null, message.getBytes());

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# Priority Queues

Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10);
channel.queueDeclare("priority.queue", true, false, false, args);

// 发送高优先级消息
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .priority(9)
    .build();
channel.basicPublish("", "priority.queue", props, message.getBytes());

1
2
3
4
5
6
7
8
9

# Batch Consumption

// 设置 QoS，每次预取 100 条消息
channel.basicQos(100);

// 批量确认
channel.basicAck(deliveryTag, true);  // multiple=true 确认所有小于等于该 deliveryTag 的消息

1
2
3
4
5

# Best Practices

# Connection Management

Use connection pool: Avoid frequently creating and closing connections
Set reasonable heartbeat: Prevent connection staleness caused by network anomalies
Exception handling: Catch exceptions and perform reconnection

# Message Reliability

Use manual acknowledgments: Ensure messages are acknowledged only after correct processing
Message persistence: Set delivery_mode=2 for important messages
Queue persistence: Set durable=true for important queues
Use Confirm mode: Ensure messages are successfully sent to the exchange

# Performance Optimization

Batch operations: Batch send messages and batch acknowledgments
Set reasonable QoS: Prevent consumers from accumulating too many messages
Use multiple channels: Create multiple channels on a single connection for concurrent processing
Choose the right queue type: Quorum Queue for high availability, Stream Queue for high throughput

# Exception Handling

// 连接断开的回调
factory.setExceptionHandler(new DefaultExceptionHandler() {
    @Override
    public void handleConnectionRecoveryException(Connection conn, Throwable exception) {
        // 记录日志并进行重连
    }
});

// 自动恢复
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);  // 5秒重试间隔
factory.setTopologyRecoveryEnabled(true);  // 自动恢复拓扑结构

1
2
3
4
5
6
7
8
9
10
11
12

编辑页面

#开发手册

← Core Engine User Manual Migration Manual→

Development Manual

# Preface

# Target Audience

# Related Documents

# Technical Support

# Development Preparation

# Connection Parameters

# Client Dependencies

# Java

# Python

# Go

# Node.js

# Java Development Example

# Producer

# Consumer

# Transactional Messages

# Confirm Mode

# Python Development Example

# Producer

# Consumer

# Go Development Example

# Producer

# Consumer

# Node.js Development Example

# Producer

# Consumer

# Advanced Features

# Delayed Messages

# Priority Queues

# Batch Consumption

# Best Practices

# Connection Management

# Message Reliability

# Performance Optimization

# Exception Handling