Message Queue for RocketMQ
Message Queue for RocketMQ
Document
- Documentation
Message Queue for RocketMQGetting StartedProducing and consuming normal messages
Copy page
Download PDFGetting Started
Producing and consuming normal messages
Copy pageDownload PDFProducing and consuming normal messages
Once you have set up your development environment and created the necessary resources, you can use the open-source RocketMQ SDK via TCP to produce and receive normal messages. This topic uses the Java SDK to demonstrate the basic steps for sending and receiving normal messages with Message Queue for RocketMQ.
Notes
When using the Java SDK to connect to Message Queue for RocketMQ for producing or consuming messages, you must configure the appropriate parameters. Learn more in Parameter description. Message Queue for RocketMQ provides code examples for common production and consumption scenarios. The comments in the code examples explain each key parameter. For more information, see SDK Reference.
Prerequisites
- You have configured your development environment and created the necessary resources, including a RocketMQ instance. For instructions, see Preparing your environment and Creating resources.
- To access your RocketMQ instance over the Internet, you must enable public access to your instance and apply for an Elastic IP Address (EIP) in the same region. We recommend that the EIP's bandwidth limit exceed your estimated peak public traffic. For instructions, see Configuring public access and SSL authentication.
Download the SDK
You can use the Java SDK for RocketMQ to connect to Message Queue for RocketMQ. Version 4.8.0 is recommended. For more information, see Java SDK download page.
Adding dependencies via Maven
Add the following dependencies to your pom.xml file.
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-acl</artifactId>
<version>4.8.0</version>
</dependency>
Sending normal messages
You can send normal messages in the following ways:
- From the Topic Management tab of the Message Queue for RocketMQ console, select the topic you want to test and click Send message in the Operation column. For more information, see Online debugging.
- We recommend using this method only for debugging. Sending messages through the console during normal operations can create unwanted test data.
- After downloading and installing the Java SDK for RocketMQ, run the sample code below to send normal messages asynchronously.
After sending a message, you can use the message query feature to confirm that the message has been delivered to the server and is available for retrieval. For instructions on querying messages, see Message query.
import java.util.Date;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
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;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
public class RocketMQAsyncProducer {
// Provide your group ID (GroupID shown in the Message Queue for RocketMQ console), as well as your access key ID and access key secret.
private static RPCHook getAclRPCHook() {
return new AclClientRPCHook(new SessionCredentials("YOUR ACCESS KEY", "YOUR SECRET KEY"));
}
public static void main(String[] args) throws MQClientException {
/**
* Create a producer.
* Provide the access point information, which can be found in the Message Queue for RocketMQ console. For example, the VPC access point (TCP) is http://{INSTANCE_ID}.rocketmq.ivolces.com:9876.
* Provide your access key ID and access key secret.
*/
DefaultMQProducer producer = new DefaultMQProducer("YOUR GROUP ID", getAclRPCHook());
producer.setNamesrvAddr("YOUR ACCESS POINT");
// If the SSL authentication policy only allows SSL connections, you must set setUseTLS(true) when accessing your instance over the Internet.
//producer.setUseTLS(true);
producer.start();
for (int i = 0; i < 128; i++) {
try {
Message msg = new Message("YOUR TOPIC",
"YOUR MESSAGE TAG",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
producer.send(msg, new SendCallback() {
public void onSuccess(SendResult result) {
// Message sent successfully.
System.out.println("send message success. msgId= " + result.getMsgId());
}
public void onException(Throwable throwable) {
// Message sending failed. Please retry. You can resend the message or save it for later retry.
System.out.println("send message failed.");
throwable.printStackTrace();
}
});
} catch (Exception e) {
// Message sending failed. Please retry. You can resend the message or save it for later retry.
System.out.println(new Date() + " Send mq message failed.");
e.printStackTrace();
}
}
// Shut down the producer before the application exits.
// Note: It is not mandatory to shut down the producer.
producer.shutdown();
}
}
Subscribing to normal messages
After successfully sending messages, start a consumer client to subscribe to them. Code example:
import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
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.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
public class RocketMQPushConsumer {
// Provide your group ID (GroupID shown in the Message Queue for RocketMQ console), as well as your access key ID and access key secret.
private static RPCHook getAclRPCHook() {
return new AclClientRPCHook(new SessionCredentials("YOUR ACCESS KEY", "YOUR SECRET KEY"));
}
public static void main(String[] args) throws MQClientException {
// Provide your group ID (GroupID shown in the Message Queue for RocketMQ console), as well as your access key ID and access key secret.
// Provide the access point for your RocketMQ instance.
// Provide the topic you created via the Message Queue for RocketMQ console.
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR GROUP ID", getAclRPCHook(), new AllocateMessageQueueAveragely());
consumer.setNamesrvAddr("http://{INSTANCE_ID}.rocketmq.volces.com.com:9876");
consumer.subscribe("YOUR TOPIC", "*");
// If the SSL authentication policy only allows SSL connections, you must set setUseTLS(true) when accessing your instance over the Internet.
//consumer.setUseTLS(true);
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("Receive New Messages: %s %n", msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
}
}
Last updated: 2026.08.28 10:19:42
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
