人人狠狠综合久久888亚洲,久久人妻无码中交,每日更新国产AV网址

本文以Java SDK為例介紹如何在VPC環境下使用SDK接入云消息隊列 Kafka 版的默認接入點并收發消息。

前提條件

步驟三：創建資源
安裝1.8或以上版本JDK
安裝2.5或以上版本Maven
安裝編譯工具。
本文以IntelliJ IDEA Ultimate為例。

安裝Java依賴庫

在IDEA中創建一個Java工程。
在pom.xml中添加以下依賴。
```
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.0</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.6</version>
</dependency>
```
說明建議您保持服務端和客戶端版本一致，即保持客戶端庫版本和云消息隊列 Kafka 版實例的大版本一致。您可以在云消息隊列 Kafka 版控制臺的實例詳情頁面獲取云消息隊列 Kafka 版實例的大版本。

準備配置

可選：下載SSL根證書。如果是SSL接入點，需下載該證書。
訪問Aliware-kafka-demos，單擊，下載Demo工程到本地并解壓。
在解壓的Demo工程中，找到kafka-java-demo文件夾，將此文件導入IntelliJ IDEA。
可選：如果是SSL接入點或者SASL接入點實例，修改配置文件kafka_client_jaas.conf。關于實例不同接入點的信息，請參見接入點對比。
```
KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="xxxx"
  password="xxxx";
}; 
```
username和password為實例的用戶名和密碼。
- 如果實例未開啟ACL，您可以在云消息隊列 Kafka 版控制臺的實例詳情頁面配置信息區域獲取默認用戶的用戶名和密碼。
- 如果實例已開啟ACL，請確保要使用的SASL用戶為PLAIN類型且已授權收發消息的權限，詳情請參見SASL用戶授權。

修改配置文件kafka.properties。

##==============================通用配置參數==============================
bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx
topic=xxx
group.id=xxx
##=======================以下內容請根據實際情況配置========================
##SSL接入點配置
ssl.truststore.location=/xxxx/kafka.client.truststore.jks
java.security.auth.login.config=/xxxx/kafka_client_jaas.conf
##SASL接入點PLAIN機制配置
java.security.auth.login.config.plain=/xxxx/kafka_client_jaas_plain.conf
##SASL接入點SCRAM機制配置
java.security.auth.login.config.scram=/xxxx/kafka_client_jaas_scram.conf


參數	描述
bootstrap.servers	接入點信息。您可在云消息隊列 Kafka 版控制臺的實例詳情頁面的接入點信息區域獲取。
topic	實例的Topic名稱。您可在云消息隊列 Kafka 版控制臺的Topic 管理頁面獲取。
group.id	實例的Group。您可在云消息隊列 Kafka 版控制臺的Group 管理頁面獲取。說明如果應用運行producer.go發送消息，該參數可以不配置；如果應用運行consumer.go訂閱消息，該參數必須配置。
ssl.truststore.location	SSL根證書的路徑。其中xxxx需要根據實際情況替換。例如，/home/doc/project/kafka-java-demo/ssl/src/main/resources/kafka.client.truststore.jks。說明如果是默認接入點實例或者SASL接入點實例，該參數不需要配置；如果是SSL接入點實例，該參數必須配置。
kafka_client_jaas.conf	JAAS配置文件所在路徑，其中xxxx需要根據實際情況替換。例如，/home/doc/project/kafka-java-demo/ssl/src/main/resources/kafka_client_jaas_scram.conf。說明如果是默認接入點實例，該參數不需要配置；如果是SSL接入點實例或者SASL接入點實例，該參數必須配置。

發送消息

編譯并運行KafkaProducerDemo.java發送消息。

示例代碼

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
//如果是SSL接入點實例或者SASL接入點實例，請注釋以下第一行代碼。
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/*
*如果是SSL接入點實例或者SASL接入點實例，請取消注釋以下兩行代碼。
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaProducerDemo {

    public static void main(String args[]) {
          
       /*
        * 如果是SSL接入點實例，請取消注釋以下一行代碼。
        設置JAAS配置文件的路徑。
        JavaKafkaConfigurer.configureSasl();
        */
         
       /*
        * 如果是SASL接入點PLAIN機制實例，請取消注釋以下一行代碼。
        設置JAAS配置文件的路徑。
        JavaKafkaConfigurer.configureSaslPlain();
        */
       
       /*
        * 如果是SASL接入點SCRAM機制實例，請取消注釋以下一行代碼。
        設置JAAS配置文件的路徑。
        JavaKafkaConfigurer.configureSaslScram();
        */

        //加載kafka.properties。
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        //設置接入點，請通過控制臺獲取對應Topic的接入點。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * 如果是SSL接入點實例，請取消注釋以下四行代碼。
        * 與sasl路徑類似，該文件也不能被打包到jar中。
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        * 根證書store的密碼，保持不變。
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        * 接入協議，目前支持使用SASL_SSL協議接入。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        * SASL鑒權方式，保持不變。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * 如果是SASL接入點PLAIN機制實例，請取消注釋以下兩行代碼。
        * 接入協議。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * Plain方式。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * 如果是SASL接入點SCRAM機制實例，請取消注釋以下兩行代碼。
        * 接入協議。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * SCRAM方式。
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
        */

        //云消息隊列 Kafka 版消息的序列化方式。
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //請求的最長等待時間。
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        //設置客戶端內部重試次數。
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        //設置客戶端內部重試間隔。
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
         
       /*
        * 如果是SSL接入點實例或，請取消注釋以下一行代碼。
        * Hostname校驗改成空。
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        */

        //構造Producer對象，注意，該對象是線程安全的，一般來說，一個進程內一個Producer對象即可。
        //如果想提高性能，可以多構造幾個對象，但不要太多，最好不要超過5個。
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        //構造一個云消息隊列 Kafka 版消息。
        String topic = kafkaProperties.getProperty("topic"); //消息所屬的Topic，請在控制臺申請之后，填寫在這里。
        String value = "this is the message's value"; //消息的內容。

        try {
            //批量獲取Future對象可以加快速度，但注意，批量不要太大。
            List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
            for (int i =0; i < 100; i++) {
                //發送消息，并獲得一個Future對象。
                ProducerRecord<String, String> kafkaMessage =  new ProducerRecord<String, String>(topic, value + ": " + i);
                Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
                futures.add(metadataFuture);

            }
            producer.flush();
            for (Future<RecordMetadata> future: futures) {
                //同步獲得Future對象的結果。
                try {
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("Produce ok:" + recordMetadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            //客戶端內部重試之后，仍然發送失敗，業務要應對此類錯誤。
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }
}

訂閱消息

選擇以下任意一種方式訂閱消息。

單Consumer訂閱消息：編譯并運行KafkaConsumerDemo.java發送消息。

示例代碼

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* 如果是SSL接入點實例，請取消注釋以下三行代碼；如果是SASL接入點，請取消注釋以下前兩行代碼。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaConsumerDemo {

    public static void main(String args[]) {

        //設置JAAS配置文件的路徑。
        /*
         * 如果是SSL接入點實例，請取消注釋以下一行代碼。
        JavaKafkaConfigurer.configureSasl();
         */
                        
        /*
         * 如果是SASL接入點PLAIN機制實例，請取消注釋以下一行代碼。
        JavaKafkaConfigurer.configureSaslPlain();
         */
                        
        /*
        * 如果是SASL接入點SCRAM機制實例，請取消注釋以下一行代碼。
        JavaKafkaConfigurer.configureSaslScram();
        */

        //加載kafka.properties
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        //設置接入點，請通過控制臺獲取對應Topic的接入點。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

        //如果是SSL接入點實例，請注釋以下第一行代碼。
        //可更加實際拉去數據和客戶的版本等設置此值，默認30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * 如果是SSL接入點實例，請取消注釋以下六行代碼。
         * 設置SSL根證書的路徑，請記得將XXX修改為自己的路徑。
         * 與SASL路徑類似，該文件也不能被打包到jar中。
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * 根證書存儲的密碼，保持不變。
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * 接入協議，目前支持使用SASL_SSL協議接入。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * SASL鑒權方式，保持不變。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * 兩次Poll之間的最大允許間隔。
         * 消費者超過該值沒有返回心跳，服務端判斷消費者處于非存活狀態，服務端將消費者從Group移除并觸發Rebalance，默認30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         * 設置單次拉取的量，走公網訪問時，該參數會有較大影響。
         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
         */

        //如果是SASL接入點PLAIN機制實例，請注釋以下一行代碼。
       //可更加實際拉去數據和客戶的版本等設置此值，默認30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * 如果是SASL接入點PLAIN機制實例，請取消注釋以下三行代碼。
         * 接入協議。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * Plain方式。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * 兩次Poll之間的最大允許間隔。
         * 消費者超過該值沒有返回心跳，服務端判斷消費者處于非存活狀態，服務端將消費者從Group移除并觸發Rebalance，默認30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        //如果是SASL接入點SCRAM機制實例，請注釋以下一行代碼。
       //可更加實際拉去數據和客戶的版本等設置此值，默認30s
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * 如果是SASL接入點SCRAM機制實例，請取消注釋以下四行代碼。
         * 接入協議。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * SCRAM方式。
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         * 兩次Poll之間的最大允許間隔。
         * 消費者超過該值沒有返回心跳，服務端判斷消費者處于非存活狀態，服務端將消費者從Group移除并觸發Rebalance，默認30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        //每次poll的最大數量。
        //注意該值不要改得太大，如果poll太多數據，而不能在下次poll之前消費完，則會觸發一次負載均衡，產生卡頓。
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        //消息的反序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //當前消費實例所屬的消費組，請在控制臺申請之后填寫。
        //屬于同一個組的消費實例，會負載消費消息。
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
        
        //如果是SSL接入點實例，請取消注釋以下一行代碼。
        //Hostname校驗改成空。
        //props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        //構造消息對象，也即生成一個消費實例。
        KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
        //設置消費組訂閱的Topic，可以訂閱多個。
        //如果GROUP_ID_CONFIG是一樣，則訂閱的Topic也建議設置成一樣。
        List<String> subscribedTopics =  new ArrayList<String>();
        
        //如果是SSL接入點實例，請注釋以下前五行代碼，取消注釋第六行代碼。
        //如果需要訂閱多個Topic，則在這里add進去即可。
        //每個Topic需要先在控制臺進行創建。
        String topicStr = kafkaProperties.getProperty("topic");
        String[] topics = topicStr.split(",");
        for (String topic: topics) {
            subscribedTopics.add(topic.trim());
        }
        //subscribedTopics.add(kafkaProperties.getProperty("topic"));
        consumer.subscribe(subscribedTopics);

        //循環消費消息。
        while (true){
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                //必須在下次poll之前消費完這些數據, 且總耗時不得超過SESSION_TIMEOUT_MS_CONFIG。
                //建議開一個單獨的線程池來消費消息，然后異步返回結果。
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Throwable ignore) {

                }
          
                e.printStackTrace();
            }
        }
    }
}

多Consumer訂閱消息：編譯并運行KafkaMultiConsumerDemo.java消費消息。

示例代碼

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
//如果是SSL接入點實例或者SASL接入點實例，請取消注釋以下第一行代碼。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* 如果是SSL接入點實例，請取消注釋以下前三行代碼；如果是SASL接入點實例，請取消注釋以下前兩行代碼。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/
import org.apache.kafka.common.errors.WakeupException;

/**
 * 本教程演示如何在一個進程內開啟多個Consumer同時消費Topic。
 * 注意全局Consumer數量不要超過訂閱的Topic總分區數。
 */
public class KafkaMultiConsumerDemo {

    public static void main(String args[]) throws InterruptedException {
        
        //設置JAAS配置文件的路徑。
        /* 
         * 如果是SSL接入點實例，請取消注釋以下一行代碼。
         JavaKafkaConfigurer.configureSasl();
         */
                            
        /* 
         * 如果是SASL接入點PLAIN機制實例，請取消注釋以下一行代碼。
         JavaKafkaConfigurer.configureSaslPlain(); 
         */
                            
        /* 
         * 如果是SASL接入點SCRAM機制實例，請取消注釋以下一行代碼。
         JavaKafkaConfigurer.configureSaslScram();
         */


        //加載kafka.properties。
        Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        //設置接入點，請通過控制臺獲取對應Topic的接入點。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        
        /*
         * 如果是SSL接入點實例，請取消注釋以下四行代碼。
         * 與SASL路徑類似，該文件也不能被打包到JAR中。
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * 根證書存儲的密碼，保持不變。
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * 接入協議，目前支持使用SASL_SSL協議接入。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * SASL鑒權方式，保持不變。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */
        
        /*
         * 如果是SASL接入點PLAIN機制實例，請取消注釋以下兩行代碼。
         * 接入協議。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * Plain方式。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */

        /* 
         * 如果是SASL接入點SCRAM機制實例，請取消注釋以下兩行代碼。
         * 接入協議。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * SCRAM方式。
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         */

        //兩次Poll之間的最大允許間隔。
        //消費者超過該值沒有返回心跳，服務端判斷消費者處于非存活狀態，服務端將消費者從Group移除并觸發Rebalance，默認30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        //每次Poll的最大數量。
        //注意該值不要改得太大，如果Poll太多數據，而不能在下次Poll之前消費完，則會觸發一次負載均衡，產生卡頓。
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        //消息的反序列化方式。
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //當前消費實例所屬的消費組，請在控制臺申請之后填寫。
        //屬于同一個組的消費實例，會負載消費消息。
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

        /* 
         * 如果是SSL接入點實例，請取消注釋以下一行代碼。
         * 構造消費對象，也即生成一個消費實例。
         * Hostname校驗改成空。
         props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
         */

        int consumerNum = 2;
        Thread[] consumerThreads = new Thread[consumerNum];
        for (int i = 0; i < consumerNum; i++) {
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

            List<String> subscribedTopics = new ArrayList<String>();
            subscribedTopics.add(kafkaProperties.getProperty("topic"));
            consumer.subscribe(subscribedTopics);

            KafkaConsumerRunner kafkaConsumerRunner = new KafkaConsumerRunner(consumer);
            consumerThreads[i] = new Thread(kafkaConsumerRunner);
        }

        for (int i = 0; i < consumerNum; i++) {
            consumerThreads[i].start();
        }

        for (int i = 0; i < consumerNum; i++) {
            consumerThreads[i].join();
        }
    }

    static class KafkaConsumerRunner implements Runnable {
        private final AtomicBoolean closed = new AtomicBoolean(false);
        private final KafkaConsumer consumer;

        KafkaConsumerRunner(KafkaConsumer consumer) {
            this.consumer = consumer;
        }

        @Override
        public void run() {
            try {
                while (!closed.get()) {
                    try {
                        ConsumerRecords<String, String> records = consumer.poll(1000);
                        //必須在下次Poll之前消費完這些數據, 且總耗時不得超過SESSION_TIMEOUT_MS_CONFIG。
                        for (ConsumerRecord<String, String> record : records) {
                            System.out.println(String.format("Thread:%s Consume partition:%d offset:%d", Thread.currentThread().getName(), record.partition(), record.offset()));
                        }
                    } catch (Exception e) {
                        try {
                            Thread.sleep(1000);
                        } catch (Throwable ignore) {

                        }
                        e.printStackTrace();
                    }
                }
            } catch (WakeupException e) {
                //如果關閉則忽略異常。
                if (!closed.get()) {
                    throw e;
                }
            } finally {
                consumer.close();
            }
        }
        //可以被另一個線程調用的關閉Hook。
        public void shutdown() {
            closed.set(true);
            consumer.wakeup();
        }
    }
}

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

默認接入點收發消息

前提條件

安裝Java依賴庫

準備配置

發送消息

訂閱消息