全部产品
Search
文档中心

云消息队列 Kafka 版:使用实例接入点收发消息

更新时间:Feb 01, 2024

当使用应用程序与Kafka集群建立连接时,可通过实例接入点接入云消息队列 Kafka 版并收发消息。Kafka实例提供默认接入点、SSL接入点和SASL接入点,以满足您不同的连接和安全需求。默认接入点适用于在保密性较高的VPC环境收发消息;SASL接入点适用于无需对传输链路加密但需对消息收发鉴权;同时需要链路加密和消息鉴权,建议使用SSL接入点。

环境准备

  • 安装1.8或以上版本JDK。具体操作。请参见安装JDK

  • 安装2.5或以上版本Maven。具体操作,请参见安装Maven

  • 安装编译工具。

    本文以IntelliJ IDEA Ultimate为例。

  • 购买并部署云消息队列 Kafka 版实例,如果实例类型为VPC实例,则仅显示默认接入点。如果实例类型为公网/VPC实例,则同时显示默认接入点和SSL接入点。实例默认不开启SASL接入点,因此默认不显示SASL接入点。如需使用SASL接入点,您需要手动开启。具体操作,请参见SASL用户授权

    • 默认接入点:适用于在VPC环境收发消息,但不支持SASL校验。

    • SASL接入点:适用于在VPC环境收发消息,且支持SASL校验。

    • SSL接入点:适用于在公网环境收发消息,且支持SASL校验。

安装Java依赖库

使用Java SDK接入Kafka实例的依赖如下所示。此依赖已在下文提到的kafka-java-demo文件夹中的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 版实例的大版本。

准备配置

  1. 可选:下载SSL根证书。如果是SSL接入点,需下载该证书。

  2. 访问Aliware-kafka-demos,单击download,下载Demo工程到本地并解压。

  3. 在解压的Demo工程中,找到kafka-java-demo文件夹,将此文件夹导入IntelliJ IDEA。

  4. 可选:如果采用SSL接入点或者SASL接入点接入实例,需修改配置文件kafka_client_jaas.conf。关于实例不同接入点的信息,请参见接入点对比

    KafkaClient {
      org.apache.kafka.common.security.plain.PlainLoginModule required
      username="xxxx"
      password="xxxx";
    }; 

    当实例类型为VPC实例时,只有在同一VPC内的资源才能访问Kafka实例,数据传输具有较高的安全性和私密性。在安全要求更高的场景下,可以开启ACL功能,消息需结合SASL身份验证鉴权之后再在安全通道传输。您可以根据消息传输的安全要求级别选择PLAIN机制或SCRAM机制进行身份认证。更多信息,请参见开启ACL

    当实例类型为公网/VPC实例时,公网环境必须对消息进行鉴权与加密,SASL的PLAIN机制必须与SSL一起用作传输层,采用SASL_SSL协议,确保消息在没有加密的情况下不会在线路上传输明文。

    示例中的usernamepassword为实例的SASL用户名和密码。

    • 如果实例未开启ACL,对于已开启公网的实例,您可以在云消息队列 Kafka 版控制台实例详情页面配置信息区域获取默认用户的用户名和密码。

    • 如果实例已开启ACL,请确保要使用的SASL用户为PLAIN类型且已授权收发消息的权限,详情请参见SASL用户授权

  5. 修改配置文件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根证书的路径。请将上文准备配置中第一步下载的SSL证书文件保存至本地路径,用本地路径替换示例代码中的xxxx。例如:/home/ssl/kafka.client.truststore.jks

    重要

    如果是默认接入点实例或者SASL接入点实例,该参数不需要配置;如果是SSL接入点实例,该参数必须配置。

    java.security.auth.login.config

    JAAS配置文件所在路径,将Demo工程中的kafka_client_jaas.conf文件另存至本地路径,用本地路径替换示例代码中的xxxx。例如:/home/ssl/kafka_client_jaas.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();
        }
    }
}

常见问题

云消息队列 Kafka 版的SASL_SSL证书怎么配置?

访问上文准备配置的第一步中的地址,下载SSL证书到本地路径,配置Demo工程中kafka.properties配置文件的ssl.truststore.location参数,即将SSL证书本地路径配置在此参数中。

使用Java SDK通过实例接入点收发消息时,可以绑定自己的SSL证书吗?

不支持。建议您使用云消息队列 Kafka 版提供的SSL证书。

相关文档