亚洲国产初高中生女av,欧美在线成人怡红院,尹人久久久香蕉精品

阿里云的云消息隊列 RabbitMQ 版支持AMQP 0-9-1協議，兼容開源的RabbitMQ客戶端，您可以使用開源的客戶端SDK接入云消息隊列 RabbitMQ 版服務端進行消息收發。

前提條件

背景信息

借助訪問控制RAM的RAM用戶，您可以實現阿里云賬號（主賬號）和RAM用戶（子賬號）權限分割，按需為RAM用戶賦予不同的權限，并避免因暴露阿里云賬號密鑰而造成安全風險。

收發消息流程（以Java語言為例）

說明

云消息隊列 RabbitMQ 版與開源RabbitMQ完全兼容。更多語言SDK，請參見開源RabbitMQ AMQP協議支持的多語言或框架SDK。

獲取接入點

您需要在云消息隊列 RabbitMQ 版控制臺獲取實例的接入點。在收發消息時，您需要為發布端和訂閱端配置該接入點，通過接入點接入云消息隊列 RabbitMQ 版實例。

登錄云消息隊列 RabbitMQ 版控制臺，然后在左側導航欄選擇實例列表。
在實例列表頁面的頂部菜單欄選擇地域，然后在實例列表中，單擊目標實例名稱。

在實例詳情頁面的接入點信息頁簽，將鼠標指針移動到目標類型的接入點，單擊該接入點右側的圖標，復制該接入點。

類型	說明	示例值
公網接入點	公網環境可讀寫。按量付費實例默認支持，預付費實例需在購買時選擇才支持。	XXX.mq-amqp.cn-hangzhou-a.aliyuncs.com
VPC接入點	VPC環境可讀寫。按量付費實例和預付費實例默認都支持。	XXX.mq-amqp.cn-hangzhou-a-internal.aliyuncs.com

安裝Java依賴庫

在pom.xml添加以下依賴。

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.5.0</version> <!-- 支持開源所有版本 -->
</dependency>

生成用戶名密碼

開源RabbitMQ客戶端接入云上服務時，需要先通過AccessKey ID和AccessKey Secret生成用戶名和密碼，將用戶名和密碼設置到開源客戶端SDK的userName和passWord參數中。云消息隊列 RabbitMQ 版會通過用戶名和密碼進行權限認證。

登錄云消息隊列 RabbitMQ 版控制臺，然后在左側導航欄選擇實例列表。
在實例列表頁面的頂部菜單欄選擇地域，然后在實例列表中，單擊目標實例名稱。
在左側導航欄，單擊靜態用戶名密碼。
在靜態用戶名密碼頁面，單擊創建用戶名密碼。
在創建用戶名密碼面板，輸入AccessKey ID和AccessKey Secret，然后單擊確定。
說明
AccessKey ID和AccessKey Secret需要在阿里云RAM控制臺獲取，具體獲取方式，請參見創建AccessKey。
靜態用戶名密碼頁面，顯示創建的靜態用戶名與密碼，密碼處于隱藏狀態。
在創建的靜態用戶名密碼的密碼列，單擊顯示密碼，可查看用戶名的密碼。

創建客戶端連接

創建連接管理工廠ConnectionFactory.java，用于啟動開源客戶端和云消息隊列 RabbitMQ 版服務端的連接。


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 {
        //create a new con
        Connection con = createCon();

        //create a new channel
        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);

        //設置為true，開啟Connection自動恢復功能；設置為false，關閉Connection自動恢復功能。
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setVirtualHost(virtualHost);
        // 默認端口。
        factory.setPort(port);

        if (enableSSL) {
            setSSL(factory);
        }

        // 基于網絡環境合理設置超時時間。
        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) {
            }
        }
    }

創建生產者發送消息

創建并編譯運行Producer.java。

重要

編譯運行Producer.java生產消息之前，您需要根據代碼提示信息配置參數列表中所列舉的參數。

表 1. 參數列表

參數	示例值	描述
hostName	1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com	云消息隊列 RabbitMQ 版實例的接入點。獲取方式，請參見獲取接入點。
port	5672	非加密端口：5672。默認為非加密端口。加密端口：5671。
userName	MjoxODgwNzcwODY5MD****	云消息隊列 RabbitMQ 版實例的靜態用戶名，用于權限驗證。獲取方式，請參見生成用戶名密碼。
passWord	NDAxREVDQzI2MjA0OT****	云消息隊列 RabbitMQ 版實例的靜態用戶名密碼，用于權限驗證。獲取方式，請參見生成用戶名密碼。
virtualHost	vhost_test	云消息隊列 RabbitMQ 版實例的Vhost，需要提前創建。具體操作，請參見前提條件。
exchangeName	ExchangeTest	云消息隊列 RabbitMQ 版的Exchange。您可以直接在代碼中設置一個Exchange名稱，系統會自動創建對應的Exchange。您也可以在云消息隊列 RabbitMQ 版控制臺創建Exchange。具體操作，請參見創建Exchange。
bindingKey	BindingKeyTest	云消息隊列 RabbitMQ 版Exchange與Queue綁定關系的Binding Key。您可以直接在代碼中設置一個BindingKey，系統會自動創建Binding Key對應的綁定關系。您也可以在云消息隊列 RabbitMQ 版控制臺創建綁定關系。具體操作，請參見綁定Queue。
queueName	QueueTest	云消息隊列 RabbitMQ 版的Queue。您可以直接在代碼中設置一個Queue，系統會自動創建對應的Queue。您也可以在云消息隊列 RabbitMQ 版控制臺創建Queue。具體操作，請參見創建Queue。

import com.rabbitmq.client.AMQP;
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 {
    //設置為云消息隊列 RabbitMQ 版實例的接入點。
    public static final String hostName = "1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com";
    //設置為云消息隊列 RabbitMQ 版實例的靜態用戶名。
    public static final String userName = "MjoxODgwNzcwODY5MD****";
    //設置為云消息隊列 RabbitMQ 版實例的靜態用戶名密碼。
    public static final String password = "NDAxREVDQzI2MjA0OT****";
    //設置為云消息隊列 RabbitMQ 版實例的Vhost名稱。
    public static final String virtualHost = "vhost_test";

    //如果使用5671端口，需要enableSSL設置為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 bindingKey;

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

    public static void main(String[] args) throws IOException, TimeoutException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        //構建連接工廠。
        ConnectionFactory factory = new ConnectionFactory(hostName, port, userName, password, virtualHost, enableSSL);

        //初始化生產者。
        Producer producer = new Producer(factory, "ExchangeTest", "QueueTest", "BindingKeyTest");

        //declare。
        producer.declare();

        producer.initChannel();

        //發送消息。
        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, bindingKey);
    }
    

    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, bindingKey, 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;
        }
    }
}

說明

云消息隊列 RabbitMQ 版會對單實例的TPS流量峰值進行限流，更多信息，請參見實例限流最佳實踐。

創建消費者訂閱消息

創建并編譯運行Consumer.java。

重要

編譯運行Consumer.java訂閱消息之前，您需要根據代碼提示信息配置參數列表中所列舉的參數。


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 {
    //設置為云消息隊列 RabbitMQ 版實例的接入點。
    public static final String hostName = "1880770****.mq-amqp.cn-hangzhou-a.aliyuncs.com";
    //設置為云消息隊列 RabbitMQ 版實例的靜態用戶名。
    public static final String userName = "MjoxODgwNzcwODY5MD****";
    //設置為云消息隊列 RabbitMQ 版實例的靜態用戶名密碼。
    public static final String password = "NDAxREVDQzI2MjA0OT****";
    //設置為云消息隊列 RabbitMQ 版實例的Vhost名稱。
    public static final String virtualHost = "vhost_test";
    
    //如果使用5671端口，需要enableSSL設置為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);
         
        //設置為云消息隊列 RabbitMQ 版實例的Queue名稱。需要和生產者中設置的Queue名稱一致。
        Consumer consumer = new Consumer(channel, "queue-1");

        consumer.consume();
    }

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

                //業務處理。
                System.out.println("receive: msgId=" + properties.getMessageId());

                //消費者需要在有效時間內提交ack，否則消息會重新推送，最多推送16次。
                //若推送16次還未成功，則消息被丟棄或者進入死信Exchange。
                //專業版實例的有效時間為1分鐘，企業版和Serverless實例為5分鐘，鉑金版實例為30分鐘。
                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();
    }
}

說明

云消息隊列 RabbitMQ 版與開源RabbitMQ完全兼容。更多參數說明，請參見開源RabbitMQ客戶端文檔。

結果驗證

您可以在云消息隊列 RabbitMQ 版控制臺通過消息查詢或軌跡查詢驗證消息的收發狀態和消息軌跡。具體操作，請參見

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

主賬號和子賬號授權場景