Message Queue for RabbitMQ
Message Queue for RabbitMQ
Document
- Documentation
Message Queue for RabbitMQSDK referenceJava SDK (AMQP protocol)
Copy page
Download PDFSDK reference
Java SDK (AMQP protocol)
Copy pageDownload PDFJava SDK (AMQP protocol)
This topic uses the Client SDK for Java as an example to describe the process of sending and receiving messages using an open-source SDK with the AMQP protocol, along with sample code. The process of sending and receiving messages is similar for SDKs in other languages or frameworks.
Notes
- This document provides sample code to demonstrate message production and consumption under the AMQP protocol. Copy the AMQP endpoint from the console according to your protocol type to configure the relevant parameters.
- Use a persistent connection for sending and receiving messages. Avoid creating a new connection for each message, as this can cause unnecessary resource consumption and may trigger SYN Flood protection on the server.
- Directly using the administrator user for production and consumption may cause information leakage risks. Before producing and consuming, create a new non-Admin user on the RabbitMQ cluster management Web UI and bind the user to a Virtual Host. For operation steps, see Bind user and Vhost.
Example projects
For sample code sending and receiving messages with RabbitMQ, see the official RabbitMQ documentation RabbitMQ SDK samples and the code projects for common scenarios in GitHub under rabbitmq-tutorials.
Step 1: Environment preparation
- Install JDK 8 or a later version in the client environment.
- Add dependencies.
- If the project uses Maven, add the following dependency to the pom.xml file.
- <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.12.0</version></dependency>
- If the project uses Gradle, add the following dependency:
- compile 'com.rabbitmq:amqp-client:5.12.0'
Step 2: Access configuration
When using the Client to connect to a RabbitMQ instance for message production and consumption, configure the following parameters in your code. Make sure all parameters are set correctly in the client code.
Step 3: Message production
The sample code for message production is as follows:
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;
import java.util.HashMap;
import java.util.UUID;
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyManagementException {
// Use the intranet endpoint domain name when SSL authentication is disabled. Use the public network endpoint domain name when SSL authentication is enabled.
final String host = "rbtmq-xxxxx.rabbitmq.volces.com";
// The intranet endpoint uses port 5672, and the public network endpoint uses port 5671.
final int port = 5671;
// It is recommended to use a user with a role other than Admin.
final String username = "demo-user";
final String password = "demo-user-password";
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(host);
factory.setPort(port);
factory.setUsername(username);
factory.setPassword(password);
// To send and receive messages using SSL, remove the comment below.
//factory.useSslProtocol();
// Configure connection parameters. Adjust based on your business needs and network conditions.
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);
factory.setConnectionTimeout(30 * 1000);
factory.setHandshakeTimeout(30 * 1000);
factory.setShutdownTimeout(0);
// The following configures objects related to the AMQP 0-9-1 protocol.
final String vhost = "/";
final String exchange = "demo-exchange";
final String exchangeType = "direct";
final String queue = "demo-queue";
final String bindingKey = "foo";
// Set the Vhost name. Ensure it is created in the RabbitMQ cluster web console.
factory.setVirtualHost(vhost);
// Create the Connection and Channel.
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// Declare a direct type Exchange. If already created manually in the console, skip this step.
channel.exchangeDeclare(exchange, exchangeType, true, false, false, null);
HashMap<String, Object> arg = new HashMap<>();
channel.queueDeclare(queue, true, false, false, arg);
channel.queueBind(queue, exchange, bindingKey);
// Start message production.
for (int i = 0; i < 100; i++ ) {
// Fill in the appropriate BindingKey according to your business requirements.
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(UUID.randomUUID().toString()).build();
channel.basicPublish(exchange, bindingKey, true, props,
("demo message " + i).getBytes(StandardCharsets.UTF_8));
}
channel.close();
connection.close();
}
}
To ensure successful message writing before returning, use the Publish Confirm mechanism. Two implementation methods are available: synchronous and asynchronous. Choose one based on your business requirements.
// Synchronous mode.
channel.confirmSelect();
for (int i = 0; i < 100; i++) {
channel.basicPublish(exchange, key, null, body);
}
channel.waitForConfirms(5000); // Wait for all messages in this batch to be confirmed
// Asynchronous mode
channel.confirmSelect();
// Use an ordered set to record message sequence numbers pending confirmation
ConcurrentNavigableMap<Long, String> outstanding = new ConcurrentSkipListMap<>();
channel.addConfirmListener(
(deliveryTag, multiple) -> { // Callback for successful confirmation
if (multiple) {
outstanding.headMap(deliveryTag, true).clear(); // Remove all records with sequence numbers less than or equal to deliveryTag
} else {
outstanding.remove(deliveryTag);
}
},
(deliveryTag, multiple) -> { // Callback for confirmation failure. You can implement resend or alarm logic here
log.error("NACK deliveryTag={}, multiple={}", deliveryTag, multiple);
}
);
long tag = channel.getNextPublishSeqNo(); // Get the message sequence number before sending
outstanding.put(tag, body);
channel.basicPublish(exchange, key, null, body.getBytes());
Step 4: Message consumption
warning
When using the AMQP 0-9-1 protocol, retrieving messages one by one with the basic.get method is not recommended, as this is less efficient than consumers that remain online. For more details, refer to the RabbitMQ official documentation.
The sample code for message consumption is as follows.
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;
public class Consumer {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException, NoSuchAlgorithmException, KeyManagementException {
// Use the private network access point domain name when SSL authentication is not enabled; use the public network access point domain name when SSL authentication is enabled.
final String host = "rbtmq-xxxxx.rabbitmq.volces.com";
// The private network access point uses port 5672; the public network access point uses port 5671.
final int port = 5671;
// It is recommended to use users with non-Admin roles.
final String username = "demo-user";
final String password = "demo-user-password";
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(host);
factory.setPort(port);
factory.setUsername(username);
factory.setPassword(password);
// To send and receive messages using SSL, remove the comment below.
// factory.useSslProtocol();
// Configure connection parameters and adjust according to your business needs and network conditions.
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);
factory.setConnectionTimeout(30 * 1000);
factory.setHandshakeTimeout(30 * 1000);
factory.setShutdownTimeout(0);
// The following are configuration objects relevant to the AMQP 0-9-1 protocol.
final String vhost = "/";
final String exchange = "demo-exchange";
final String exchangeType = "direct";
final String queue = "demo-queue";
final String bindingKey = "foo";
// Set the Vhost name. Ensure it has already been created in the RabbitMQ cluster web console.
factory.setVirtualHost(vhost);
// Create the Connection and Channel.
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// Declare a direct type Exchange. If you have already manually created it in the console, skip this step.
channel.exchangeDeclare(exchange, exchangeType, true, false, false, null);
HashMap<String, Object> arg = new HashMap<>();
channel.queueDeclare(queue, true, false, false, arg);
channel.queueBind(queue, exchange, bindingKey);
// Start consuming messages.
channel.basicConsume(queue, false, "consumertag", new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
System.out.printf("receive message: " + new String(body, StandardCharsets.UTF_8) + ", deliveryTag: " + envelope.getDeliveryTag() + ", messageId: " + properties.getMessageId());
// Manually ACK each message.
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
CountDownLatch latch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
try {
channel.getConnection().close();
} catch (IOException e) {
System.out.println("close connection error." + e);
}
latch.countDown();
}));
latch.await();
connection.close();
}
}
Error codes (AMQP)
Last updated: 2026.08.31 18:14:59
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
