kafka java API的使用

　　Kafka包含四種核心的API:

　　1、Producer API支持應用將數據流發送到Kafka集群的主題

　　2、Consumer API支持應用從Kafka集群的主題中讀取數據流

　　3、Streams API支持數據流從輸入主題轉化到輸出主題

　　4、Connect API支持實現持續地從一些源系統或應用划入Kafka或者從Kafka推入一些源系統或應用的接口。

　　我們這里主要討論Producer API和Consumer API的使用，由於最新版的kafka java api中使用了一些jdk8的新特性，所以要求我們在本機上jdk版本要在8以上。

　　pom.xml如下：

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

Producer API

　　Producer用來向Kafka集群中發布消息記錄的Kafka客戶端。Producer是線程安全的，並且通常來講，在多個線程間共享一個producer要比每個線程都創建一個producer速度更快。producer代碼示例：

package com.example.demo;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class MyProducer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.1.124:9092");
                props.put("acks", "all");
                props.put("retries", 0);
                props.put("batch.size", 16384);
                props.put("linger.ms", 1);
                props.put("partitioner.class", "com.example.demo.MyPartitioner");
                props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<String, String>("powerTopic", Integer.toString(i), Integer.toString(i)));

        producer.close();

    }
}

　　properties里用到的配置參數在kafka的源碼里org.apache.kafka.clients.producer.ProducerConfig類中，這里說一下常用的：

　　bootstrap.servers 配置項處需要填寫我們要發送到的Kafka集群地址。

　　ack 配置項用來控制producer要求leader確認多少消息后返回調用成功。當值為0時producer不需要等待任何確認消息。當值為1時只需要等待leader確認。當值為-1或all時需要全部ISR集合返回確認才可以返回成功。

　　retries 當 retries > 0 時，如果發送失敗，會自動嘗試重新發送數據。發送次數為retries設置的值。

　　buffer.memory、batch.size、linger.ms三個參數用來控制緩沖區大小和延遲發送時間，具體含義可以參考官方文檔的配置。

　　key.serializer 和 value.serializer 指定使用什么序列化方式將用戶提供的key和value進行序列化。

Consumer API

　　Consumer的API分為High-level API和Low-level API。前者提供了高度抽象的API，使用起來簡單、方便。因此本文將主要講述High-level API。Low-level API提供了更強的控制能力，但使用起來較為繁瑣。自動提交consumer代碼示例：

package com.example.demo;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class MyAutoCommitConsumer {

    public static void main(String[] args) {
         Properties props = new Properties();
         props.put("bootstrap.servers", "192.168.1.124:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");
         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList("powerTopic"));
         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records)
                 System.out.printf("partition = %d,offset = %d, key = %s, value = %s%n",record.partition(), record.offset(), record.key(), record.value());
         }
    }
}

　　properties里用到的配置參數在kafka的源碼里org.apache.kafka.clients.consumer.ConsumerConfig類中，本例中用到參數解釋如下：

　　bootstrap.servers配置項指定了consumer需要連接的服務器集群。多台服務器用“,”分隔

　　enable.auto.commit配置項指定了提交offset的方式為自動提交，auto.commit.interval.ms配置項配置了每次自動提交的時間間隔。

　　group.id 即消費者組標簽，本例中消費者組的名稱為test。

　　key.deserializer和value.deserializer指用什么方式進行反序列化。

　　自動提交offset的方式非常簡單，但多數情況下，我們不會使用自動提交的方式。因為不論從Kafka集群中拉取的數據是否被處理成功，offset都會被更新，也就是如果處理過程中出現錯誤可能會出現數據丟失的情況。所以多數情況下我們會選擇手動提交方式，我們看到 enable.auto.commit 配置項被設置為false，代表手動提交。示例代碼如下：

package com.example.demo;

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

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class MyManualCommitConsumer {

    public static void main(String[] args) {
         Properties props = new Properties();
         props.put("bootstrap.servers", "192.168.1.124:9092");
         props.put("group.id", "test");
         props.put("enable.auto.commit", "false");
         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         @SuppressWarnings("resource")
         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList("myFirstTopic"));
         final int minBatchSize = 200;
         List<ConsumerRecord<String, String>> list = new ArrayList<>();
         while (true) {
             ConsumerRecords<String, String> records = consumer.poll(100);
             for (ConsumerRecord<String, String> record : records) {
                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                 list.add(record);
             }
             if (list.size() >= minBatchSize) {
                 System.out.println("list中的緩存數據大於minBatchSize時批量進行處理");
                 consumer.commitSync();
                 System.out.println("全部數據處理成功后手動提交");
                 list.clear();
             }
         }

    }

}

　　另外需注意，consumer是有狀態的，所以不是線程安全的，所以在進行多線程操作時需要在每個線程實例化一個consumer。