ApsaraMQ for RabbitMQ:Send and receive messages using an SDK

Obtain an instance endpoint

Specify the endpoint that producers and consumers use to access the ApsaraMQ for RabbitMQ instance.

1. Log on to the ApsaraMQ for RabbitMQ console. In the left-side navigation pane, click Instances.

2. In the top navigation bar of the Instances page, select the region where the instance that you want to manage resides. Then, in the instance list, click the name of the instance that you want to manage.

3. On the Endpoint Information tab of the Instance Details page, hover over the target endpoint and copy it by clicking the icon next to it.

   Type	Description	Example
   Public endpoint	Access the instance over the Internet. Pay-as-you-go instances support public endpoints by default. Subscription instances require you to enable Internet access during creation.	XXX.net.mq.amqp.aliyuncs.com
   VPC endpoint	Access the instance from a VPC. Both pay-as-you-go and subscription instances support VPC endpoints by default.	XXX.vpc.mq.amqp.aliyuncs.com

Add the Java dependency

1. Create a Java project in IntelliJ IDEA.

2. Add the following dependency to the pom.xml file:

   <dependency>
       <groupId>com.rabbitmq</groupId>
       <artifactId>amqp-client</artifactId>
       <version>5.5.0</version> <!-- All versions of open-source RabbitMQ are supported. -->
   </dependency>

Create a username and password

The open-source RabbitMQ SDK requires a username and password for authentication. Generate credentials using one of the following methods, and configure the userName and passWord parameters in your code.

• Open-source authentication and authorization

  1. Log on to the ApsaraMQ for RabbitMQ console. In the left-side navigation pane, click Instances.

  2. In the top navigation bar of the Instances page, select the region where the instance that you want to manage resides. Then, in the instance list, click the name of the instance that you want to manage.

  3. In the left-side navigation pane, click Users and Permissions.

  4. On the Users and Permissions page, click Create Username/Password.

  5. In the Create Username/Password panel, complete the Username, Password, and Confirm Password fields, and then click OK.

  Note

  After creating the credentials, grant permissions to the user. Manage users and permissions.

• Resource Access Management (RAM)

  1. Log on to the ApsaraMQ for RabbitMQ console. In the left-side navigation pane, click Instances.

  2. In the top navigation bar of the Instances page, select the region where the instance that you want to manage resides. Then, in the instance list, click the name of the instance that you want to manage.

  3. In the left-side navigation pane, click Users and Permissions.

  4. On the Users and Permissions page, click Create Username/Password.

  5. In the Create Username/Password panel, enter the AccessKey ID and AccessKey Secret, and then click OK.

     Note

     Obtain the AccessKey ID and AccessKey Secret from the RAM console. Create an AccessKey.

     The new credentials appear on the Users and Permissions page. The password is hidden.

  6. In the Password column for the created credentials, click Display to view the password.

Step 4: Create the connection factory

The ConnectionFactory class handles connection setup, SSL/TLS configuration, and automatic recovery. Create a file named ConnectionFactory.java:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;

public class ConnectionFactory {
    private final String hostName;
    private final int port;
    private final String userName;
    private final String password;
    private final String virtualHost;
    private final boolean enableSSL;

    public ConnectionFactory(String hostName, int port, String userName,
                            String password, String virtualHost, boolean enableSSL) {
        this.hostName = hostName;
        this.port = port;
        this.userName = userName;
        this.password = password;
        this.virtualHost = virtualHost;
        this.enableSSL = enableSSL;
    }

    public Channel createChannel() throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        Connection con = createCon();
        return con.createChannel();
    }

    private Connection createCon() throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();

        factory.setHost(hostName);
        factory.setUsername(userName);
        factory.setPassword(password);

        // Reconnect automatically if the connection drops.
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setVirtualHost(virtualHost);
        factory.setPort(port);

        if (enableSSL) {
            setSSL(factory);
        }

        // Adjust these timeouts based on your network conditions.
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);
        factory.setShutdownTimeout(0);

        return factory.newConnection();
    }

    private void setSSL(com.rabbitmq.client.ConnectionFactory factory) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
        TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        trustManagerFactory.init((KeyStore) null);
        sslContext.init(null, trustManagerFactory.getTrustManagers(), null);
        factory.useSslProtocol(sslContext);
    }

    public void closeCon(Channel channel) {
        if (channel != null && channel.getConnection() != null) {
            try {
                channel.getConnection().close();
            } catch (Throwable t) {
            }
        }
    }
}

Key configuration:

Produce messages

In your Java project, create a producer program named Producer.java, configure the parameters based on the description in SDK parameters, and run the program. For precautions about sending messages, see What do I need to pay attention to when I produce messages?

Sample code:

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class Producer {
    // The endpoint of the ApsaraMQ for RabbitMQ instance. 
    public static final String hostName = "1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com";
    // The static username of the ApsaraMQ for RabbitMQ instance. 
    public static final String userName = "MjoxODgwNzcwODY5MD****";
    // The static password of the ApsaraMQ for RabbitMQ instance. 
    public static final String password = "NDAxREVDQzI2MjA0OT****";
    // The name of the vhost of the ApsaraMQ for RabbitMQ instance. 
    public static final String virtualHost = "vhost_test";

    // If you want to use port 5671, you must set the enableSSL parameter to true. 
    public static final int port = 5672;
    public static final boolean enableSSL = false;

    private Channel channel;
    private final ConcurrentNavigableMap<Long/*deliveryTag*/, String/*msgId*/> outstandingConfirms;
    private final ConnectionFactory factory;
    private final String exchangeName;
    private final String queueName;
    private final String routingKey;

    public Producer(ConnectionFactory factory, String exchangeName, String queueName, String routingKey) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        this.factory = factory;
        this.outstandingConfirms = new ConcurrentSkipListMap<>();
        this.channel = factory.createChannel();
        this.exchangeName = exchangeName;
        this.queueName = queueName;
        this.routingKey = routingKey;
    }

    public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        // Create a connection factory. 
        ConnectionFactory factory = new ConnectionFactory(hostName, port, userName, password, virtualHost, enableSSL);

        // Initialize the producer. 
        Producer producer = new Producer(factory, "ExchangeTest", "QueueTest", "RoutingKeyTest");

        // Declare the producer. 
        producer.declare();

        producer.initChannel();

        // Send messages. 
        producer.doSend("hello,amqp");
    }

    private void initChannel() throws IOException {
        channel.confirmSelect();

        ConfirmCallback cleanOutstandingConfirms = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);

                for (Long tag : confirmed.keySet()) {
                    String msgId = confirmed.get(tag);
                    System.out.format("Message with msgId %s has been ack-ed. deliveryTag: %d, multiple: %b%n", msgId, tag, true);
                }

                confirmed.clear();
            } else {
                String msgId = outstandingConfirms.remove(deliveryTag);
                System.out.format("Message with msgId %s has been ack-ed. deliveryTag: %d, multiple: %b%n", msgId, deliveryTag, false);
            }
        };
        channel.addConfirmListener(cleanOutstandingConfirms, (deliveryTag, multiple) -> {
            String msgId = outstandingConfirms.get(deliveryTag);
            System.err.format("Message with msgId %s has been nack-ed. deliveryTag: %d, multiple: %b%n", msgId, deliveryTag, multiple);
            // send msg failed, re-publish
        });


        channel.addReturnListener(returnMessage -> System.out.println("return msgId=" + returnMessage.getProperties().getMessageId()));
    }

    private void declare() throws IOException {
        channel.exchangeDeclare(exchangeName, "direct", true);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
    }
    

    private void doSend(String content) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        try {
            String msgId = UUID.randomUUID().toString();
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();

            channel.basicPublish(exchangeName, routingKey, true, props, content.getBytes(StandardCharsets.UTF_8));

            outstandingConfirms.put(channel.getNextPublishSeqNo(), msgId);
        } catch (AlreadyClosedException e) {
            //need reconnect if channel is closed.
            String message = e.getMessage();

            System.out.println(message);

            if (channelClosedByServer(message)) {
                factory.closeCon(channel);
                channel = factory.createChannel();
                this.initChannel();
                doSend(content);
            } else {
                throw e;
            }
        }
    }

    private boolean channelClosedByServer(String errorMsg) {
        if (errorMsg != null
            && errorMsg.contains("channel.close")
            && errorMsg.contains("reply-code=541")
            && errorMsg.contains("reply-text=InternalError")) {
            return true;
        } else {
            return false;
        }
    }
}

Subscribe to messages

In your Java project, create a consumer program named Consumer.java, configure the parameters based on the description in SDK parameters, and run the program.

Sample code:

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;

public class Consumer {
    // The endpoint of the ApsaraMQ for RabbitMQ instance. 
    public static final String hostName = "1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com";
    // The static username of the ApsaraMQ for RabbitMQ instance. 
    public static final String userName = "MjoxODgwNzcwODY5MD****";
    // The static password of the ApsaraMQ for RabbitMQ instance. 
    public static final String password = "NDAxREVDQzI2MjA0OT****";
    // The name of the vhost of the ApsaraMQ for RabbitMQ instance. 
    public static final String virtualHost = "vhost_test";
    
    // If you want to use port 5671, you must set the enableSSL parameter to true. 
    public static final int port = 5672;
    public static final boolean enableSSL = false;

    private final Channel channel;
    private final String queue;

    public Consumer(Channel channel, String queue) {
        this.channel = channel;
        this.queue = queue;
    }

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        ConnectionFactory factory = new ConnectionFactory(hostName, port, userName, password, virtualHost, enableSSL);
        Channel channel = factory.createChannel();
        channel.basicQos(50);
         
        // The name of the queue on the ApsaraMQ for RabbitMQ instance. The queue name that you specify for this parameter must be consistent with the queue name that you specified when you created the producer. 
        Consumer consumer = new Consumer(channel, "queue-1");

        consumer.consume();
    }

    public void consume() throws IOException, InterruptedException {
        channel.basicConsume(queue, false, new DefaultConsumer(channel) {
            @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {

                // Process the message. 
                System.out.println("receive: msgId=" + properties.getMessageId());

                // The consumer must commit acknowledgment (ACK) within the validity period. Otherwise, the message is pushed again. The message can be pushed up to 16 times. 
                // If the message still fails to be pushed after 16 times, it is discarded or sent to the dead-letter exchange. 
                // The validity period is 1 minute for Professional Edition instances, 5 minutes for Enterprise and Serverless Edition instances, and 30 minutes for Enterprise Platinum Edition instances. 
                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();
    }
}

SDK parameters

References

• ApsaraMQ for RabbitMQ is fully compatible with open-source RabbitMQ and supports SDKs for multiple languages. For information about SDKs for other programming languages, see Language or framework SDK Java RabbitMQ Java Client Library Spring Framework Spring AMQP project for Java .NET .NET SDK Python Python SDK PHP PHP SDK Rust Rust SDK C and C++ C and C++ SDK Go Go SDK JavaScript and Node JavaScript and Node SDK Objective-C and Swift Objective-C and Swift SDK. For other parameters, see Client Documentation.

• To troubleshoot client errors, refer to AMQP error codes.

• Query messages and traces in the ApsaraMQ for RabbitMQ console to verify message delivery. For more information, see Messages and Message traces.