Kafka client會消費kafka cluster中的記錄。

它將透明地處理Kafka叢集中的伺服器故障，並透明地適應它在叢集內遷移的資料分割槽。該客戶機還與伺服器互動，以允許使用者組使用消費者組來負載平衡消費(如下所述)。

偏移量

消費者組和訂閱主題

用法示例

1. 自動確認Offset

Properties props = new Properties();
/* 定義kakfa 服務的地址，不需要將所有broker指定上 */
props.put("bootstrap.servers", "localhost:9092");
/* 制定consumer group */
props.put("group.id", "test");
/* 是否自動確認offset */
props.put("enable.auto.commit", "true");
/* 自動確認offset的時間間隔 */
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
/* key的序列化類 */
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
/* value的序列化類 */
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 /* 定義consumer */
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
/* 消費者訂閱的topic, 可同時訂閱多個 */
consumer.subscribe(Arrays.asList("foo", "bar"));

 /* 讀取資料，讀取超時時間為100ms */
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
        System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
}

說明: 
1. bootstrap.servers 只是代表kafka的連線入口，只需要指定叢集中的某一broker； 
2. 一旦consumer和kakfa叢集建立連線，consumer會以心跳的方式來高速叢集自己還活著，如果session.timeout.ms 內心跳未到達伺服器，伺服器認為心跳丟失，會做rebalence。

2. 手工控制Offset

如果consumer在獲得資料後需要加入處理，資料完畢後才確認offset，需要程式來控制offset的確認。舉個栗子： 
consumer獲得資料後，需要將資料持久化到DB中。自動確認offset的情況下，如果資料從kafka叢集讀出，就確認，但是持久化過程失敗，就會導致資料丟失。我們就需要控制offset的確認。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
/* 關閉自動確認選項 */
props.put("enable.auto.commit", "false");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
final int minBatchSize = 200;
List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        buffer.add(record);
    }
    /* 資料達到批量要求，就寫入DB，同步確認offset */
    if (buffer.size() >= minBatchSize) {
        insertIntoDb(buffer);
        consumer.commitSync();
        buffer.clear();
    }
}

還可以精細的控制對具體分割槽具體offset資料的確認：

try {
    while(running) {
        ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
            for (ConsumerRecord<String, String> record : partitionRecords) {
                System.out.println(record.offset() + ": " + record.value());
            }
            /* 同步確認某個分割槽的特定offset */
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
            consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
        }
    }
} finally {
  consumer.close();
}

說明：確認的offset為已接受資料最大offset+1。

3. 分割槽訂閱

可以向特定的分割槽訂閱訊息。但是會失去partion的負載分擔。有幾種場景可能會這麼玩： 
1. 只需要獲取本機磁碟的分割槽資料； 
2. 程式自己或者外部程式能夠自己實現負載和錯誤處理。例如YARN/Mesos的介入，當consumer掛掉後，再啟動一個consumer。

String topic = "foo";
TopicPartition partition0 = new TopicPartition(topic, 0);
TopicPartition partition1 = new TopicPartition(topic, 1);
consumer.assign(Arrays.asList(partition0, partition1));

說明： 
1. 此種情況用了consumer Group，也不會做負載均衡。 
2. topic的訂閱和分割槽訂閱不可以在同一consumer中混用。

4. 外部儲存offset

消費者可以自定義kafka的offset儲存位置。該設計的主要目的是讓消費者將資料和offset進行原子性的儲存。這樣可以避免上面提到的重複消費問題。舉慄說明： 
訂閱特定分割槽。儲存所獲得的記錄時，將每條記錄的offset一起儲存。保證資料和offset的儲存是原子性的。當非同步儲存被異常打斷時，凡已經儲存的資料，都有有相應的offset記錄。這種方式可以保證不會有資料丟失，也不會重複的從服務端讀取。 
如何配置實現： 
1. 去使能offset自動確認：enable.auto.commit=false； 
2. 從ConsumerRecord中獲取offset，儲存下來; 
3. Consumer重啟時，呼叫seek(TopicPartition, long)重置在服務端的消費記錄。

如果消費分割槽也是自定義的，這種方式用起來會很爽。如果分割槽是自動分配的，當分割槽發生reblance的時候，就要考慮清楚了。如果因為升級等原因，分割槽漂移到一個不會更新offset的consumer上，那就日了狗了。 
該情況下： 
1. 原consumer需要監聽分割槽撤銷事件，並在撤銷時確認好offset。介面:ConsumerRebalanceListener.onPartitionsRevoked(Collection)； 
2. 新consumer監聽分割槽分配事件，獲取當前分割槽消費的offset。介面：ConsumerRebalanceListener.onPartitionsAssigned(Collection)； 
3. consumer監聽到 ConsumerRebalance事件，還沒有處理或者持久化的快取資料flush掉。

5. 控制消費位置

大多數情況下，服務端的Consumer的消費位置都是由客戶端間歇性的確認。Kafka允許Consumer自己設定消費起點，達到的效果： 
1. 可以消費已經消費過的資料； 
2. 可以跳躍性的消費資料； 
看下這樣做的一些場景： 
1. 對Consumer來說，資料具備時效性，只需要獲取最近一段時間內的資料，就可以進行跳躍性的獲取資料； 
2. 上面自己存offset的場景，重啟後就需要從指定的位置開始消費。 
介面上面已經提到過了，用seek(TopicPartition, long)。、 
麻蛋，說指標不就好了，這一小節就是多餘的叨叨。

6. 控制消費流Consumption Flow Control

如果一個consumer同時消費多個分割槽，預設情況下，這多個分割槽的優先順序是一樣的，同時消費。Kafka提供機制，可以讓暫停某些分割槽的消費，先獲取其他分割槽的內容。場景舉慄： 
1. 流式計算，consumer同時消費兩個Topic，然後對兩個Topic的資料做Join操作。但是這兩個Topic裡面的資料產生速率差距較大。Consumer就需要控制下獲取邏輯，先獲取慢的Topic，慢的讀到資料後再去讀快的。 
2. 同樣多個Topic同時消費，但是Consumer啟動是，本地已經存有了大量某些Topic資料。此時就可以優先去消費下其他的Topic。

調控的手段：讓某個分割槽消費先暫停，時機到了再恢復，然後接著poll。介面：pause(TopicPartition…)，resume(TopicPartition…)

7. 多執行緒處理模型 Multi-threaded Processing

Kafka的Consumer的介面為非執行緒安全的。多執行緒共用IO，Consumer執行緒需要自己做好執行緒同步。 
如果想立即終止consumer，唯一辦法是用呼叫介面：wakeup()，使處理執行緒產生WakeupException。

public class KafkaConsumerRunner implements Runnable {
    /* 注意，這倆貨是類成員變數 */
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final KafkaConsumer consumer;

    public void run() {
        try {
            consumer.subscribe(Arrays.asList("topic"));
            while (!closed.get()) {
                ConsumerRecords records = consumer.poll(10000);
                // Handle new records
            }
        } catch (WakeupException e) {
            // Ignore exception if closing
            if (!closed.get()) throw e;
        } finally {
            consumer.close();
        }
    }

    // Shutdown hook which can be called from a separate thread
    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }
}

說明： 
1. KafkaConsumerRunner是runnable的，請自覺補腦多執行緒執行； 
2. 外部執行緒控制KafkaConsumerRunner執行緒的停止； 
3. 主要說的是多執行緒消費同一topic，而不是消費同一分割槽；

比較一下兩種模型：

Consumer單執行緒模型

優點：實現容易； 
沒有執行緒之間的協作。通常比下面的那種更快； 
單分割槽資料的順序處理； 
缺點：多個TCP連線，但是關係不大，kafka對自己的server自信滿滿； 
太多的Request可能導致server的吞吐降低一丟丟； 
consumer數量受到分割槽數量限制，一個consumer一個分割槽；

Consumer多執行緒模型

優點：一個consumer任意多的執行緒，執行緒數不用受到分割槽數限制； 
缺點：如果有保序需求，自己要加控制邏輯； 
該模型中如果手動offset，自己要加控制邏輯； 
一種可行的解決辦法：為每個分割槽分配獨立的儲存，獲取的資料根據資料所在分割槽進行hash儲存。這樣可以解決順序消費，和offset的確認問題。