All Products
Search
Document Center

ApsaraMQ for RabbitMQ:JMS P2P (RAM role for cross-account authorization)

Last Updated:Sep 08, 2023

This topic describes how to connect a Java Message Service (JMS) client to ApsaraMQ for RabbitMQ and send and receive messages in point-to-point (P2P) messaging pattern in scenarios in which a Resource Access Management (RAM) role is used to grant permissions across Alibaba Cloud accounts.

Prerequisites

Background information

  • JMS P2P

    The JMS P2P messaging model has the following features:

    • Each message has only one consumer. After a producer sends a message to a queue, a specified consumer registered for the queue can receive the message. JMS P2P model

    • Producers and consumers have no timing dependencies. Producers can send messages to queues regardless of whether consumers are active. Consumers can receive messages from queues regardless of whether producers are active.

  • If you need to use the temporary Security Token Service (STS) token generated for a RAM role to access ApsaraMQ for RabbitMQ, you must set the AccessKey ID, AccessKey Secret, and SecurityToken parameters in the AliyunCredentialsProvider class for permission verification.

    Important

    If you use an SDK to send or receive messages on your client, we recommend that you use persistent connections. This way, the client does not need to establish connections every time you use the client to send or receive messages. Frequent creation of connections consumes a large number of network and broker resources and may even trigger protection against SYN flood attacks on the broker. For more information, see Connection.

Messaging process

JMS P2P

Obtain an endpoint

You must obtain the endpoint that is used to access your instance in the ApsaraMQ for RabbitMQ console. Before you send and receive messages, you must specify the endpoint in the code of the producer and consumer. Then, the producer and consumer can access the ApsaraMQ for RabbitMQ instance by using the endpoint.

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

  2. In the top navigation bar of the Instances page, select a region. 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, move the pointer over the type of endpoint that you want to use. Then, click the Copy icon on the right side of the endpoint to copy the endpoint.

    Type

    Description

    Example

    Public endpoint

    You can access an instance over the Internet to read and write data. By default, pay-as-you-go instances support public endpoints. To use a public endpoint for a subscription instance, you must select a public endpoint when you create the subscription instance.

    XXX.mq-amqp.cn-hangzhou-a.aliyuncs.com

    VPC endpoint

    You can access an instance over a virtual private cloud (VPC) to read and write data. By default, pay-as-you-go and subscription instances support VPC endpoints.

    XXX.mq-amqp.cn-hangzhou-a-internal.aliyuncs.com

Add JMS dependencies

Add the following dependencies 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>
<dependency>
    <groupId>com.alibaba.mq-amqp.jms</groupId>
    <artifactId>mq-amqp-jms-client</artifactId>
    <version>1.11.2-1.0.0</version>
    <exclusions>
        <exclusion>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>com.alibaba.mq-amqp</groupId>
    <artifactId>mq-amqp-client</artifactId>
    <version>1.0.5</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.4</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>1.7.25</version>
    <scope>compile</scope>
</dependency>

Configure the AliyunCredentialsProvider.java class

  1. Create a .properties configuration file and configure the AccessKeyID, AccessKeySecret and SecurityToken parameters in the file. For information about how to configure these parameters, see Parameters.

    # Access Key ID.
    accessKeyId=${accessKeyId}
    # Access Key Secret.
    accessKeySecret=${accessKeySecret}
    # security temp token. (optional)
    securityToken=${securityToken}
  2. Create the AliyunCredentialsProvider.java class for authentication and configure related parameters based on code comments. For more information, see Parameters.

    import java.io.FileReader;
    import java.io.IOException;
    import java.util.Properties;
    
    import com.alibaba.mq.amqp.utils.UserUtils;
    import com.rabbitmq.client.impl.CredentialsProvider;
    import org.apache.commons.lang3.StringUtils;
    import java.security.InvalidKeyException;
    import java.security.NoSuchAlgorithmException;
    
    public class AliyunCredentialsProvider implements CredentialsProvider {
    
        /**
         * Access Key ID.
         */
        private static final String accessKeyId;
        /**
         * Access Key Secret.
         */
        private static final String accessKeySecret;
        /**
         * security temp token. (optional)
         */
        private static final String securityToken;
        /**
         * The ID of the ApsaraMQ for RabbitMQ instance. You can obtain the instance ID in the ApsaraMQ for RabbitMQ console. 
         */
        private final String instanceId;
    
    
        // The AccessKey pair of your Alibaba Cloud account can be used to access all API operations. We recommend that you use the AccessKey pair of a RAM user to access API operations and perform routine O&M. 
        // We strongly recommend that you do not save your AccessKey pair in your project code. Otherwise, your AccessKey pair may be leaked and all resources that are contained in your Alibaba Cloud account are exposed to potential security risks. 
        // In this example, the AccessKey pair is saved in the configuration file. 
        static {
            // Replace ${ConfigPath} with the path of the .properties file that you created. 
            String path = "${ConfigPath}";
            Properties properties = new Properties();
    
            try {
                properties.load(new FileReader(path));
            } catch (IOException e) {
                // Automatically handle exceptions that occur during the reading of the configuration file. 
            }
    
            accessKeyId = properties.getProperty("accessKeyId");
            accessKeySecret = properties.getProperty("accessKeySecret");
            securityToken = properties.getProperty("securityToken");
    
        }
    
        public AliyunCredentialsProvider(final String instanceId) {
            this.instanceId = instanceId;
        }
    
        @Override
        public String getUsername() {
            if(StringUtils.isNotEmpty(securityToken)) {
                return UserUtils.getUserName(accessKeyId, instanceId, securityToken);
            } else {
                return UserUtils.getUserName(accessKeyId, instanceId);
            }
        }
    
        @Override
        public String getPassword() {
            try {
                return UserUtils.getPassord(accessKeySecret);
            } catch (InvalidKeyException e) {
                //todo
            } catch (NoSuchAlgorithmException e) {
                //todo
            }
            return null;
        }
    }

Parameters

Parameter

Example

Description

hostName

1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com

The endpoint of the ApsaraMQ for RabbitMQ instance. In the ApsaraMQ for RabbitMQ console, you can obtain the endpoint on the Instance Details page.

Port

5672

The default port. Use port 5672 for non-encrypted connections and port 5671 for encrypted connections.

AccessKeyID

LTAI5tJQKnB9zVvQ****

The AccessKey ID of your Alibaba Cloud account or a RAM user within the Alibaba Cloud account. You can log on to the RAM console, create a RAM role, grant the role the AliyunAMQPFullAccess permission, obtain the Alibaba Cloud Resource Name (ARN) of the RAM role, and then call the AssumeRole operation to obtain a temporary identity to assume the role. A successful call to the AssumeRole operation returns values of the AccessKeyID, AccessKeySecret, and SecurityToken of the RAM role. For more information about the ARN of a RAM role, see RAM role overview.

AccessKeySecret

jw6MawfKOVBveRr84u****

The AccessKey secret of your Alibaba Cloud account or a RAM user within the Alibaba Cloud account.

SecurityToken

CAIS9wF1q6Ft5B2yfSjIr5fkJNPkvK5tgqeScX+ElnkMXvhvgIPO0Dz2IHhNfXZuB************

The STS token of the RAM role.

instanceId

amqp-cn-v0h1kb9nu***

The ID of the ApsaraMQ for RabbitMQ instance. In the ApsaraMQ for RabbitMQ console, you can view the instance ID on the Instance Details page. For information about how to view the ID of an instance, see View the details of an instance.

virtualHost

Test

The vhost that you created in the ApsaraMQ for RabbitMQ instance. In the ApsaraMQ for RabbitMQ console, you can view the vhost on the vhosts page. For information about how to view a vhost, see View the information about the connection that is established to a vhost.

Configure an object message

Create the Person.java class that is used to configure an object message.

import java.io.Serializable;

public class Person implements Serializable{
    
    private static final long serialVersionUID = -5809782578272943****;
    String name = null;
    int age = 0;
    String address =null;

    public Person(String name, int age, String address){
        this.name = name;
        this.age = age;
        this.address = address;
    }

    public String getName(){
        return name;
    }

    public int getAge(){
        return age;
    }

    public String getAddress(){
        return address;
    }
}

Send messages

Create, compile, and run QueueSend.java to send messages.

Important

Before you compile QueueSend.java and run it to send messages, you must configure the parameters that are described in Parameters based on the comments in the code.

import com.rabbitmq.jms.admin.RMQConnectionFactory;
import com.rabbitmq.jms.admin.RMQDestination;
import com.rabbitmq.jms.client.AmqpConnectionFactoryPostProcessor;
import javax.jms.*;
import java.io.Serializable;

public class QueueSend {

    public void sendTextMsg(Session session, String MsgContent, String name) throws JMSException {
        Destination queue = new RMQDestination(name, true, false);
        MessageProducer msgProducer = session.createProducer(queue);
        Message msg = session.createTextMessage(MsgContent);
        msgProducer.send(msg);
        System.out.println("Text message sent");
    }

    public void sendMap(Session session, MapMessage map, String name) throws JMSException {
        Destination queue = new RMQDestination(name, true, false);
        MessageProducer msgProducer = session.createProducer(queue);
        msgProducer.send(map);
        System.out.println("Map message sent");
    }

    public void sendObj(Session session, Person obj, String name) throws JMSException {
        Destination queue = new RMQDestination(name, true, false);
        MessageProducer msgProducer = session.createProducer(queue);
        ObjectMessage objMsg = session.createObjectMessage((Serializable) obj);
        MessageProducer msgPorducer = session.createProducer(queue);
        msgPorducer.send(objMsg);
        System.out.println("Object message sent");
    }

    public static void main(String[] args) throws JMSException {
        final AliyunCredentialsProvider provider = new AliyunCredentialsProvider("${InstanceID}");
        RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
        connectionFactory.setHost("xxx.xxx.aliyuncs.com");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("${VhostName}");
        connectionFactory.setAmqpConnectionFactoryPostProcessor(new AmqpConnectionFactoryPostProcessor() {
            @Override
            public void postProcess(com.rabbitmq.client.ConnectionFactory cf) {
                cf.setCredentialsProvider(provider);
                cf.setAutomaticRecoveryEnabled(true);
                cf.setNetworkRecoveryInterval(5000);
                System.out.println(cf.getPassword());
            }
        });
        Connection connection = connectionFactory.createConnection();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        QueueSend qs = new QueueSend();
        qs.sendTextMsg(session, "Send a JMS text message", "queue.msgText");
        MapMessage mapMsg = session.createMapMessage();
        mapMsg.setString("name", "Li");
        mapMsg.setBoolean("IsHero", false);
        mapMsg.setInt("age", 23);
        qs.sendMap(session, mapMsg, "queue.msgMap");
        Person person = new Person("Li", 23, "Beijing-Daxing")
        qs.sendObj(session, person, "queue.msgObj");
        session.close();
        connection.close();
    }
}

Subscribe to messages

Create, compile, and run QueueAccept.java to subscribe to messages.

Important

Before you compile QueueAccept.java and run it to subscribe to messages, you must configure the parameters that are described in Parameters based on the comments in the code.

import com.rabbitmq.jms.admin.RMQConnectionFactory;
import com.rabbitmq.jms.admin.RMQDestination;
import com.rabbitmq.jms.client.AmqpConnectionFactoryPostProcessor;
import javax.jms.*;

public class QueueAccept implements MessageListener {

    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            TextMessage text = (TextMessage) message;
            try {
                System.out.println("Content of the text message" + text.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
        if (message instanceof MapMessage) {
            MapMessage map = (MapMessage) message;
            try {
                System.out.println("Name:" + map.getString("name"));
                System.out.println("Hero:" + map.getBoolean("IsHero"));
                System.out.println("Age:" + map.getInt("age"));
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
        if (message instanceof ObjectMessage) {
            ObjectMessage objMsg = (ObjectMessage) message;
            try {
                Person person = (Person) objMsg.getObject();
                System.out.println("Username:" + person.getName() + ", Age:" + person.getAge() + ", Address:" + person.getAddress());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
        try {
            message.acknowledge();
            System.out.println("Manually acknowledge the message");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws JMSException {
        RMQConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        if (connectionFactory == null) {
            final AliyunCredentialsProvider provider = new AliyunCredentialsProvider("${InstanceID}");
            connectionFactory = new RMQConnectionFactory();
            connectionFactory.setHost("xxx.xxx.aliyuncs.com");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("${VhostName}");
            connectionFactory.setAmqpConnectionFactoryPostProcessor(new AmqpConnectionFactoryPostProcessor() {
                @Override
                public void postProcess(com.rabbitmq.client.ConnectionFactory cf) {
                    cf.setCredentialsProvider(provider);
                    cf.setAutomaticRecoveryEnabled(true);
                    cf.setNetworkRecoveryInterval(5000);
                    System.out.println(cf.getPassword());
                }
            });
        }
        if (connection == null) {
            connection = connectionFactory.createConnection();
            connection.start();
        }
        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        Destination destination = new RMQDestination("queue.msgText", true, false);
        MessageConsumer consumer = session.createConsumer(destination);
        consumer.setMessageListener(new QueueAccept());
        Destination queue1 = new RMQDestination("queue.msgMap", true, false);
        MessageConsumer consumer1 = session.createConsumer(queue1);
        consumer1.setMessageListener(new QueueAccept());
        Destination queue2 = new RMQDestination("queue.msgObj", true, false);
        MessageConsumer consumer2 = session.createConsumer(queue2);
        consumer2.setMessageListener(new QueueAccept());
    }
}