開發手冊

# 前言

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

# 適用對象

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

# 相關文件

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

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

# 技術支援

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

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

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

您的姓名
公司資訊與聯絡方式
作業系統及其版本
產品版本號
出現異常及錯誤的日誌、截圖等詳細資訊

# 開發準備

# 連線參數

參數	說明	範例
host	伺服器位址	localhost
port	埠號	5672（AMQP）/ 5671（AMQP over SSL）
username	使用者名稱	guest
password	密碼	guest
virtualHost	虛擬主機	/

# 客戶端依賴

# 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 開發範例

# 生產者

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

# 消費者

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

# 事務訊息

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 模式

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 開發範例

# 生產者

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

# 消費者

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 開發範例

# 生產者

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

# 消費者

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 開發範例

# 生產者

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

# 消費者

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

# 進階功能

# 延遲訊息

使用死信交換器 + TTL 實現延遲佇列：

// 创建死信交换机和队列
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

# 優先級佇列

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

# 批次消費

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

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

1
2
3
4
5

# 最佳實踐

# 連線管理

使用連線池：避免頻繁建立和關閉連線
合理設定心跳：防止網路異常導致連線假死
異常處理：捕獲異常並進行重新連線

# 訊息可靠性

使用手動確認：確保訊息被正確處理後才確認
訊息持久化：重要訊息設定 delivery_mode=2
佇列持久化：重要佇列設定 durable=true
使用 Confirm 模式：確保訊息成功發送到交換器

# 效能最佳化

批次操作：批次傳送訊息、批次確認
合理設定 QoS：避免消費者積壓過多訊息
使用多頻道：單個連線上建立多個頻道並行處理
選擇合適的佇列類型：Quorum Queue 適合高可用場景，Stream Queue 適合高吞吐場景

# 異常處理

// 连接断开的回调
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

编辑页面

#开发手册

← 核心引擎使用者手冊遷移手冊→