Development Manual

# Preface

This document is the development guide for Kingdee Apusic Distributed Message Queue for RocketMQ (Apusic Distributed Message Queue for RocketMQ, referred to as: ADMQ for RocketMQ), helping users quickly learn how to use ADMQ for RocketMQ for development.

# 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 RocketMQ products, please refer to the following ADMQ for RocketMQ product documentation set:

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

# Technical Support

ADMQ for RocketMQ provides comprehensive technical support services. You can obtain technical support through the following methods:

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:

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

# Installation and Deployment

# Deployment and Startup

Extract the installation package

mkdir -p /apusic
cd /apusic
# Extract the installation package
#tar zxvf ADMQ-V2.0.6-RocketMQ-{build_date}.tar.gz -C /apusic/
tar zxvf ADMQ-V2.0.6.534-RocketMQ-20260625.tar.gz -C /apusic/
cd /apusic
# It is recommended to separate the nameserver and broker component folder structures
unzip rocketmq-V5.3.4-all.zip
mv rocketmq-V5.3.4-all nameserver
cd /apusic/nameserver

unzip rocketmq-V5.3.4-all.zip
mv rocketmq-V5.3.4-all broker
cd /apusic/broker

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

Modify the configuration, update the address and port in the corresponding configuration file

vi conf/nameserver.properties
vi conf/broker.conf

1
2

Start

bin/rocketmq start nameserver
bin/rocketmq start broker

1
2

# Connection Parameters

Parameter	Description	Example
namesrvAddr	NameServer address	localhost:9876
groupName	Producer/Consumer group name	test-group
accessKey	ACL access key (optional)	test-user
secretKey	ACL secret key (optional)	test-password

# Client Development

# Client Dependencies

# Java

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.8</version> (For 5.x version, use 5.1.4)
</dependency>

1
2
3
4
5

# Go

go get github.com/apache/rocketmq-clients/golang/v2

# Python

pip install rocketmq-client-python

# Java Development Examples

# Producer

# Synchronous Send

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        
        Message msg = new Message("test-topic", "tag-a", "test-key", 
            "Hello ADMQ for RocketMQ!".getBytes("UTF-8"));
        SendResult result = producer.send(msg);
        System.out.println("SendResult: " + result);
        
        producer.shutdown();
    }
}

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

# Asynchronous Send

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        
        Message msg = new Message("test-topic", "tag-a", "test-key", 
            "Hello ADMQ for RocketMQ!".getBytes("UTF-8"));
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("Send success: " + sendResult);
            }
            
            @Override
            public void onException(Throwable e) {
                System.err.println("Send failed: " + e.getMessage());
            }
        });
        
        Thread.sleep(5000);
        producer.shutdown();
    }
}

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

# One-way Send

Message msg = new Message("test-topic", "tag-a", "One-way message".getBytes());
producer.sendOneway(msg);

1
2

# Consumer

# Push Mode Consumption

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class PushConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("test-topic", "*");
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
                    ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("Received: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        System.out.println("Consumer started");
    }
}

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

# Pull Mode Consumption

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class PullConsumer {
    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("test-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("test-topic", "*");
        consumer.start();
        
        while (true) {
            List<MessageExt> msgs = consumer.poll();
            for (MessageExt msg : msgs) {
                System.out.println("Received: " + new String(msg.getBody()));
            }
        }
    }
}

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

# Go Development Examples

# Producer

package main

import (
    "context"
    "fmt"
    "github.com/apache/rocketmq-clients/golang/v2"
)

func main() {
    producer, err := golang.NewProducer(
        golang.WithEndpoints("localhost:9876"),
        golang.WithConsumerGroup("test-group"),
    )
    if err != nil {
        panic(err)
    }
    
    err = producer.Start()
    if err != nil {
        panic(err)
    }
    defer producer.Shutdown()
    
    msg := golang.NewMessage("test-topic", []byte("Hello ADMQ for RocketMQ!"))
    msg.SetTag("tag-a")
    msg.SetKeys("test-key")
    
    resp, err := producer.Send(context.TODO(), msg)
    if err != nil {
        panic(err)
    }
    fmt.Printf("Message sent: %s\n", resp[0].MsgId)
}

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

# Consumer

package main

import (
    "context"
    "fmt"
    "github.com/apache/rocketmq-clients/golang/v2"
)

func main() {
    consumer, err := golang.NewSimpleConsumer(
        golang.WithEndpoints("localhost:9876"),
        golang.WithConsumerGroup("test-group"),
        golang.WithSubscriptionExpressions(map[string]*golang.FilterExpression{
            "test-topic": golang.SUB_ALL,
        }),
    )
    if err != nil {
        panic(err)
    }
    
    err = consumer.Start()
    if err != nil {
        panic(err)
    }
    defer consumer.Shutdown()
    
    for {
        msgs, err := consumer.Receive(context.TODO(), 1, 30)
        if err != nil {
            fmt.Printf("Receive error: %v\n", err)
            continue
        }
        
        for _, msg := range msgs {
            fmt.Printf("Received: %s\n", string(msg.GetBody()))
            err = consumer.Ack(context.TODO(), msg)
            if err != nil {
                fmt.Printf("Ack error: %v\n", err)
            }
        }
    }
}

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

# Python Development Examples

# Producer

from rocketmq.client import Producer, Message

producer = Producer('test-group')
producer.set_name_server_address('localhost:9876')
producer.start()

msg = Message('test-topic')
msg.set_tags('tag-a')
msg.set_keys('test-key')
msg.set_body('Hello ADMQ for RocketMQ!')

result = producer.send_sync(msg)
print('Message sent: {result.msg_id}')

producer.shutdown()

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

# Consumer

from rocketmq.client import PushConsumer, ConsumeStatus

def callback(msg):
    print(f'Received: {msg.body.decode("utf-8")}')
    return ConsumeStatus.CONSUME_SUCCESS

consumer = PushConsumer('test-group')
consumer.set_name_server_address('localhost:9876')
consumer.subscribe('test-topic', '*', callback)
consumer.start()

print('Consumer started')
import time
time.sleep(60)
consumer.shutdown()

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

# Advanced Features

# Ordered Messages

// Send ordered message
Message msg = new Message("order-topic", "tag-a", "order-1", "content".getBytes());
SendResult result = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);

// Consume ordered message
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Received: " + new String(msg.getBody()));
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

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

# Transactional Messages

TransactionMQProducer producer = new TransactionMQProducer("tx-group");
producer.setNamesrvAddr("localhost:9876");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // Execute local transaction
            executeLocalBusiness(msg);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // Check back local transaction status
        if (isTransactionSuccess(msg)) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
});
producer.start();

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

# Delayed Messages

Message msg = new Message("delay-topic", "tag-a", "content".getBytes());
msg.setDelayTimeLevel(3);  // Delay for 10 seconds
producer.send(msg);

1
2
3

# Batch Consumption

consumer.setConsumeMessageBatchMaxSize(10);  // Consume 10 messages at a time
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
            ConsumeConcurrentlyContext context) {
        // Process messages in batch
        for (MessageExt msg : msgs) {
            processMessage(msg);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

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

编辑页面

#Development Guide

← Core Engine User Manual Migration Manual→

Development Manual

# Preface

# Target Audience

# Related Documents

# Technical Support

# Installation and Deployment

# Deployment and Startup

# Connection Parameters

# Client Development

# Client Dependencies

# Java

# Go

# Python

# Java Development Examples

# Producer

# Synchronous Send

# Asynchronous Send

# One-way Send

# Consumer

# Push Mode Consumption

# Pull Mode Consumption

# Go Development Examples

# Producer

# Consumer

# Python Development Examples

# Producer

# Consumer

# Advanced Features

# Ordered Messages

# Transactional Messages

# Delayed Messages

# Batch Consumption