开源RabbitMQ客户端接入阿里云RabbitMQ-云消息队列 RabbitMQ 版(RabbitMQ)-阿里云帮助中心

本文以调用Java SDK为例，介绍在RAM角色跨账号授权场景，通过开源SDK实现消息收发的操作过程，帮助您更好地理解消息收发的完整过程，其他语言或框架的SDK消息收发过程相似。

前提条件

背景信息

当您需要通过RAM STS角色授权的方式访问云消息队列 RabbitMQ 版服务时，需要通过阿里云提供的权限认证类（AliyunCredentialsProvider）设置 AccessKeyID、AccessKeySecret与SecurityToken进行权限认证才能访问。

借助访问控制RAM的RAM角色，您可以跨云账号授权，使某个企业访问另一个企业的云消息队列 RabbitMQ 版。

企业A希望能专注于业务系统，仅作为云消息队列 RabbitMQ 版所有者。企业A希望可以授权企业B来操作部分业务，例如：云消息队列 RabbitMQ 版的运维、监控以及管理等。
企业A希望当企业B的员工加入或离职时，无需做任何权限变更。企业B可以进一步将企业A的资源访问权限分配给企业B的RAM用户（员工或应用），并可以精细控制其员工或应用对资源的访问和操作权限。
企业A希望如果双方合同终止，企业A随时可以撤销企业B的授权。

更多信息，请参见RAM跨云账号授权。

收发消息流程（以Java语言为例）

说明

云消息队列 RabbitMQ 版与开源RabbitMQ完全兼容。更多语言SDK，请参见开源RabbitMQ AMQP协议支持的多语言或框架SDK。

安装Java依赖库

在pom.xml中添加以下依赖。

<dependency>            
    <groupId>com.rabbitmq</groupId>            
    <artifactId>amqp-client</artifactId>            
    <version>5.5.0</version> <!-- 支持开源所有版本 -->       
</dependency>        
<dependency>          
    <groupId>com.alibaba.mq-amqp</groupId>          
    <artifactId>mq-amqp-client</artifactId>         
    <version>1.0.5</version>       
</dependency>       
<dependency>          
    <groupId>com.aliyun</groupId>          
    <artifactId>alibabacloud-sts20150401</artifactId>          
    <version>1.0.4</version>       
</dependency>

配置权限认证类（AliyunCredentialsProvider）

创建权限认证类AliyunCredentialsProvider.java，根据代码提示信息，设置相关参数。具体信息，请参见参数列表。

import com.alibaba.mq.amqp.utils.UserUtils;
import com.aliyun.auth.credentials.Credential;
import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
import com.aliyun.sdk.service.sts20150401.AsyncClient;
import com.aliyun.sdk.service.sts20150401.models.AssumeRoleRequest;
import com.aliyun.sdk.service.sts20150401.models.AssumeRoleResponseBody;
import com.rabbitmq.client.impl.CredentialsProvider;
import darabonba.core.client.ClientOverrideConfiguration;

import java.security.GeneralSecurityException;
import java.time.Instant;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 调用 STS AssumeRole 获取临时凭证，并生成云消息队列 RabbitMQ 版的用户名和密码。
 * 每次建立连接时检查凭证，距离过期不足 REFRESH_AHEAD_MS 时重新调用 AssumeRole。
 */
public class AliyunCredentialsProvider implements CredentialsProvider {
    /**
     * STS 凭证有效期（秒），不能超过 RAM 角色的最大会话时间。
     */
    private static final long DURATION_SECONDS = 3600;
    /**
     * 凭证过期前多久更新凭证并重建连接。
     */
    public static final long REFRESH_AHEAD_MS = 5 * 60 * 1000L;

    private final String instanceId;
    private final String accessKeyId;
    private final String accessKeySecret;
    private final String region;
    private final String roleArn;
    private final String roleSessionName;

    private volatile StsSession session;

    /**
     * @param instanceId      实例 ID，在云消息队列 RabbitMQ 版控制台实例详情页面查看
     * @param accessKeyId     RAM 用户的 AccessKey ID，该 RAM 用户需具备 AssumeRole 权限
     * @param accessKeySecret RAM 用户的 AccessKey Secret
     * @param region          STS 接入地域，例如 cn-hangzhou
     * @param roleArn         RAM 角色 ARN，格式为 acs:ram::<account_id>:role/<role_name>
     * @param roleSessionName 角色会话名称，会记录在操作审计中，建议按应用区分
     */
    public AliyunCredentialsProvider(String instanceId, String accessKeyId, String accessKeySecret,
                                     String region, String roleArn, String roleSessionName) {
        this.instanceId = instanceId;
        this.accessKeyId = accessKeyId;
        this.accessKeySecret = accessKeySecret;
        this.region = region;
        this.roleArn = roleArn;
        this.roleSessionName = roleSessionName;
    }

    @Override
    public String getUsername() {
        StsSession current = currentSession();
        return UserUtils.getUserName(current.accessKeyId, instanceId, current.securityToken);
    }

    @Override
    public String getPassword() {
        try {
            return UserUtils.getPassord(currentSession().accessKeySecret);
        } catch (GeneralSecurityException e) {
            throw new IllegalStateException("Failed to generate password: " + e.getMessage(), e);
        }
    }

    /**
     * 当前 STS 凭证的过期时间（毫秒时间戳）。
     */
    public long getExpiration() {
        return currentSession().expirationMillis;
    }

    private synchronized StsSession currentSession() {
        if (session == null || System.currentTimeMillis() > session.expirationMillis - REFRESH_AHEAD_MS) {
            session = assumeRole();
        }
        return session;
    }

    private StsSession assumeRole() {
        StaticCredentialProvider provider = StaticCredentialProvider.create(Credential.builder()
                .accessKeyId(accessKeyId)
                .accessKeySecret(accessKeySecret)
                .build());
        try (AsyncClient client = AsyncClient.builder()
                .region(region)
                .credentialsProvider(provider)
                .overrideConfiguration(ClientOverrideConfiguration.create()
                        // Endpoint 请参见 https://api.aliyun.com/product/Sts
                        .setEndpointOverride("sts." + region + ".aliyuncs.com"))
                .build()) {
            AssumeRoleRequest request = AssumeRoleRequest.builder()
                    .roleArn(roleArn)
                    .roleSessionName(roleSessionName)
                    .durationSeconds(DURATION_SECONDS)
                    .build();
            AssumeRoleResponseBody body = client.assumeRole(request).get(30, TimeUnit.SECONDS).getBody();
            AssumeRoleResponseBody.Credentials credentials = body.getCredentials();
            if (credentials == null) {
                throw new IllegalStateException("AssumeRole returned no credentials, requestId=" + body.getRequestId());
            }
            System.out.println("[INFO] AssumeRole succeeded, expiration: " + credentials.getExpiration());
            return new StsSession(credentials.getAccessKeyId(), credentials.getAccessKeySecret(),
                    credentials.getSecurityToken(), Instant.parse(credentials.getExpiration()).toEpochMilli());
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            throw new IllegalStateException("AssumeRole failed, roleArn=" + roleArn + ": " + e.getMessage(), e);
        }
    }

    private static final class StsSession {
        final String accessKeyId;
        final String accessKeySecret;
        final String securityToken;
        final long expirationMillis;

        StsSession(String accessKeyId, String accessKeySecret, String securityToken, long expirationMillis) {
            this.accessKeyId = accessKeyId;
            this.accessKeySecret = accessKeySecret;
            this.securityToken = securityToken;
            this.expirationMillis = expirationMillis;
        }
    }
}

表 1. 参数列表

参数	示例值	描述
hostName	1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com	云消息队列 RabbitMQ 版实例接入点。
Port	5672	默认端口。非加密端口5672，加密端口5671。
AccessKeyID	yourAccessKeyID	阿里云账号或RAM用户的AccessKey ID。您可以登录RAM访问控制台，创建RAM角色，并赋予角色AliyunAMQPFullAccess权限，获取角色的ARN，调用AssumeRole接口获取一个扮演该角色的临时身份。AssumeRole执行成功会返回RAM角色的 AccessKeyID、AccessKeySecret以及SecurityToken。角色ARN的概念，请参见RAM角色概览。
AccessKeySecret	yourAccessKeySecret	阿里云账号或RAM用户的AccessKey Secret。
region	cn-hangzhou	调用对应地域的AssumeRole接口，详情请参见AssumeRole。
roleARN	acs:ram::125xxxxxxx223:role/xxx	RAM角色的ARN。格式为`acs:ram::<account_id>:role/<role_name>`。详情请参见AssumeRole。
instanceId	amqp-cn-v0h1kb9nu***	云消息队列 RabbitMQ 版的实例ID。您可以在云消息队列 RabbitMQ 版控制台的实例详情页面查看。如何查看实例ID，请参见查看实例详情。
virtualHost	Test	云消息队列 RabbitMQ 版实例的Vhost。您可以在云消息队列 RabbitMQ 版控制台的Vhost 列表页面查看。如何查看Vhost，请参见查看Vhost连接详情。
ExchangeName	ExchangeTest	云消息队列 RabbitMQ 版的Exchange。您可以在云消息队列 RabbitMQ 版控制台的Exchange 列表页面获取。
RoutingKey	RoutingKeyTest	云消息队列 RabbitMQ 版Exchange与Queue的Routing Key。您可以在云消息队列 RabbitMQ 版控制台的Exchange 列表页面查看Exchange的绑定关系，获取Routing Key。
QueueName	QueueTest	云消息队列 RabbitMQ 版的Queue。仅在订阅消息时候需要配置，您可以在云消息队列 RabbitMQ 版控制台的Exchange 列表页面，查看Exchange的绑定关系，获取Exchange綁定的Queue。

生产消息

创建并编译运行ProducerTest.java。

重要

编译运行ProducerTest.java生产消息之前，您需要根据代码提示信息配置参数列表中所列举的参数。

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

public class ProducerTest {
    // 推荐将 AK/SK/ARN 等信息配置在环境变量中，明文保存在工程代码中会带来数据泄露风险。
    // RAM 用户的 AccessKey ID 和 AccessKey Secret，该 RAM 用户只需具备 AssumeRole 权限。
    private static final String alibabaAccessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String alibabaAccessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    // STS 接入地域，例如 cn-hangzhou。
    private static final String region = System.getenv("ALIBABA_CLOUD_REGION");
    // RAM 角色 ARN，从 RAM 控制台获取。
    private static final String roleARN = System.getenv("ALIBABA_CLOUD_ROLE_ARN");
    // 实例接入点、实例 ID，在云消息队列 RabbitMQ 版控制台实例详情页面查看。
    private static final String hostName = "xxx.xxx.aliyuncs.com";
    private static final String instanceId = "${InstanceId}";
    private static final String virtualHost = "${VirtualHost}";
    // Exchange 和 RoutingKey 必须已在控制台创建并绑定。
    private static final String exchangeName = "";
    private static final String routingKey = "";

    public static void main(String[] args) throws Exception {
        AliyunCredentialsProvider credentialsProvider = new AliyunCredentialsProvider(
                instanceId, alibabaAccessKeyId, alibabaAccessKeySecret, region, roleARN, "amqp-producer");

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostName);
        factory.setPort(5672);
        factory.setVirtualHost(virtualHost);
        factory.setCredentialsProvider(credentialsProvider);
        // 开启 Connection 自动恢复，网络断开重连时会自动使用最新的 STS 凭证。
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 记录当前连接使用的 STS 凭证的过期时间。
        long expiration = credentialsProvider.getExpiration();

        for (int i = 0; i < 3600; i++) {
            try {
                // STS 凭证过期后，已建立的连接仍会保持，但收发消息会因鉴权失败被拒绝，
                // 因此需要在过期前用新凭证建立新连接，再关闭旧连接。
                if (System.currentTimeMillis() > expiration - AliyunCredentialsProvider.REFRESH_AHEAD_MS) {
                    Connection newConnection = factory.newConnection();
                    Channel newChannel = newConnection.createChannel();
                    connection.abort(10 * 1000);
                    connection = newConnection;
                    channel = newChannel;
                    expiration = credentialsProvider.getExpiration();
                    System.out.println("[INFO] Connection rebuilt with new STS credentials.");
                }
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .messageId(UUID.randomUUID().toString()).build();
                channel.basicPublish(exchangeName, routingKey, true, props,
                        ("消息发送Body-" + i).getBytes(StandardCharsets.UTF_8));
                System.out.println("[SendResult] Message sent successfully, messageId: " + props.getMessageId());
                Thread.sleep(1000L);
            } catch (Exception e) {
                // 重建连接失败时旧连接仍可使用，下次循环会重试。
                System.out.println("[ERROR] Send fail, error: " + e.getMessage());
                Thread.sleep(5000L);
            }
        }
        connection.close();
    }
}

说明

云消息队列 RabbitMQ 版会对单实例的TPS流量峰值进行限流，更多信息，请参见实例限流最佳实践。

订阅消息

创建并编译运行ConsumerTest.java订阅消息。

重要

编译运行ConsumerTest.java订阅消息之前，您需要根据代码提示信息配置参数列表中所列举的参数。

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class ConsumerTest {
    // 推荐将 AK/SK/ARN 等信息配置在环境变量中，明文保存在工程代码中会带来数据泄露风险。
    // RAM 用户的 AccessKey ID 和 AccessKey Secret，该 RAM 用户只需具备 AssumeRole 权限。
    private static final String alibabaAccessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String alibabaAccessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    // STS 接入地域，例如 cn-hangzhou。
    private static final String region = System.getenv("ALIBABA_CLOUD_REGION");
    // RAM 角色 ARN，从 RAM 控制台获取。
    private static final String roleARN = System.getenv("ALIBABA_CLOUD_ROLE_ARN");
    // 实例接入点、实例 ID，在云消息队列 RabbitMQ 版控制台实例详情页面查看。
    private static final String hostName = "xxx.xxx.aliyuncs.com";
    private static final String instanceId = "${InstanceId}";
    private static final String virtualHost = "${VirtualHost}";
    // Queue 必须已在控制台创建。
    private static final String queueName = "";

    public static void main(String[] args) throws Exception {
        AliyunCredentialsProvider credentialsProvider = new AliyunCredentialsProvider(
                instanceId, alibabaAccessKeyId, alibabaAccessKeySecret, region, roleARN, "amqp-consumer");

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostName);
        factory.setPort(5672);
        factory.setVirtualHost(virtualHost);
        factory.setCredentialsProvider(credentialsProvider);
        // 开启 Connection 自动恢复，网络断开重连时会自动使用最新的 STS 凭证。
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setConnectionTimeout(30 * 1000);
        factory.setHandshakeTimeout(30 * 1000);

        Connection connection = factory.newConnection();
        consume(connection.createChannel(), queueName);
        // 记录当前连接使用的 STS 凭证的过期时间。
        long expiration = credentialsProvider.getExpiration();
        System.out.println("Consumer started.");

        // 为便于理解，这里用循环检测凭证是否即将过期，实际业务中可使用定时任务。
        while (true) {
            if (System.currentTimeMillis() <= expiration - AliyunCredentialsProvider.REFRESH_AHEAD_MS) {
                Thread.sleep(1000L);
                continue;
            }
            try {
                // STS 凭证过期后，已建立的连接仍会保持，但消费和确认会因鉴权失败被拒绝，
                // 因此需要在过期前用新凭证建立新连接并开始消费，再关闭旧连接。
                Connection newConnection = factory.newConnection();
                consume(newConnection.createChannel(), queueName);
                connection.abort(10 * 1000);
                connection = newConnection;
                expiration = credentialsProvider.getExpiration();
                System.out.println("[INFO] Consumer restarted with new STS credentials.");
            } catch (Exception e) {
                // 重建连接失败时旧连接仍可使用，稍后重试。
                System.out.println("[ERROR] Rebuild connection fail, error: " + e.getMessage());
                Thread.sleep(10 * 1000L);
            }
        }
    }

    private static void consume(Channel channel, String queueName) throws IOException {
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 接收到的消息，进行业务逻辑处理。
                System.out.println("Received: " + new String(body, StandardCharsets.UTF_8)
                        + ", deliveryTag: " + envelope.getDeliveryTag() + ", messageId: " + properties.getMessageId());
                getChannel().basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}

查询消息

如果您想确认消息是否成功发送至云消息队列 RabbitMQ 版，可以在云消息队列 RabbitMQ 版控制台查询消息。具体操作，请参见查询消息。