本文介紹Maven方式引入Kafka客戶端，并完成Kafka實例連接以及消息生產與消費的相關示例。如果您需要在IDE中查看Demo具體表現，請查看Java開發環境搭建。

下文所有Kafka的配置信息，如實例連接地址、Topic名稱、用戶信息等，請參考收集連接信息獲取。

Maven中引入Kafka客戶端

安裝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>

配置修改

1. 如果是SSL接入點，需在控制臺下載證書。

2. 修改kafka.properties文件。

   ##==============================通用配置參數==============================
   bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx
   topic=xxx
   group.id=xxx
   ##=======================以下內容請根據實際情況配置========================
   ##SSL接入點配置證書路徑
   ssl.truststore.location=/xxxx/ssl.client.truststore.jks
   

3. 如果不是SSL接入點，需要將代碼中SSL部分注釋掉。如果是SSL接入點，需要在代碼中修改證書密碼（ssl.client.pw.key）。

   //證書store的密碼，保持不變props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "xxxxxxx");
   

配置讀取

import java.util.Properties;
public class JavaKafkaConfigurer {
    private static Properties properties;
    public synchronized static Properties getKafkaProperties() {
        if (null != properties) {
            return properties;
        }
        //獲取配置文件kafka.properties的內容
        Properties kafkaProperties = new Properties();
        try {
            kafkaProperties.load(KafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));
        } catch (Exception e) {
            //沒加載到文件，程序要考慮退出
            e.printStackTrace();
        }
        properties = kafkaProperties;
        return kafkaProperties;
    }
}

生產消息

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
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;
import org.apache.kafka.common.config.SslConfigs;

public class KafkaProducerDemo {
    public static void main(String args[]) {
        //加載kafka.properties
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();
        Properties props = new Properties();
        //設置接入點，請通過控制臺獲取對應Topic的接入點
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        //設置SSL根證書的路徑，請記得將XXX修改為自己的路徑
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        //根證書store的密碼，保持不變
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "c24f5210");
        //接入協議，目前支持使用SSL協議接入
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
        //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.RECONNECT_BACKOFF_MS_CONFIG, 3000);
        //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 {
            //批量獲取 futures 可以加快速度, 但注意，批量不要太大
            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();
        }
    }
}

消費消息

import org.apache.kafka.clients.CommonClientConfigs;
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;
import org.apache.kafka.common.config.SslConfigs;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaConsumerDemo {
    public static void main(String args[]) {
        //加載kafka.properties
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();
        Properties props = new Properties();
        //設置接入點，請通過控制臺獲取對應Topic的接入點
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        //設置SSL根證書的路徑，請記得將XXX修改為自己的路徑
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        //根證書store的密碼，保持不變
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "c24f5210");
        //接入協議，目前支持使用SSL協議接入
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
        //兩次poll之間的最大允許間隔
        //可更加實際拉去數據和客戶的版本等設置此值，默認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);
        //每次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"));
        //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>();
        //如果需要訂閱多個Topic，則在這里add進去即可
        //每個Topic需要先在控制臺進行創建
        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();
            }
        }
    }
}