請使用0.9以後的版本：

示例程式碼

 Properties props = new Properties();
        props.put("bootstrap.servers", "kafka01:9092，kafka02: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");
        
        props.put("auto.offset.reset","earliest");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("foo", "bar"));
      try{  
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
         }
        }finally{
          consumer.close();
        }
 

1、只需要配置kafka的server groupid autocommit 序列化 autooffsetreset(其中 bootstrap.server group.id key.deserializer value.deserializer 必須指定);

2、用這些Properties構建consumer物件(KafkaConsumer還有其他構造，可以把序列化傳進去);

3、subscribe訂閱topic列表（可以用正則訂閱Pattern.compile("kafka.*")

使用正則必須指定一個listener subscribe(Pattern pattern, ConsumerRebalanceListener listener)）; 可以重寫這個介面來實現 分割槽變更時的邏輯。如果設定了enable.auto.commit = true 就不用理會這個邏輯。

4、然後迴圈poll訊息(這裡的1000是超時設定，如果沒有很多資料，也就等一秒);

5、處理訊息（列印了offset key value 這裡寫處理邏輯）。

6、關閉KafkaConsumer（可以傳一個timeout值 等待秒數 預設是30）。

引數詳解

bootstrap.server(最好用主機名不用ip kafka內部用的主機名 除非自己配置了ip)

deserializer 反序列化consumer從broker端獲取的是位元組陣列，還原回物件型別。

預設有十幾種：StringDeserializer LongDeserializer DoubleDeserializer。。

也可以自定義：定義serializer格式 建立自定義deserializer類實現Deserializer 介面 重寫邏輯

​

除了四個必傳的 bootstrap.server group.id key.deserializer value.deserializer

還有session.timeout.ms "coordinator檢測失敗的時間"

是檢測consumer掛掉的時間 為了可以及時的rebalance 預設是10秒 可以設定更小的值避免訊息延遲。

max.poll.interval.ms "consumer處理邏輯最大時間"

處理邏輯比較複雜的時候 可以設定這個值 避免造成不必要的 rebalance ，因為兩次poll時間超過了這個引數，kafka認為這個consumer已經跟不上了，會踢出組，而且不能提交offset，就會重複消費。預設是5分鐘。

auto.offset.reset "無位移或者位移越界時kafka的應對策略"

所以如果啟動了一個group從頭消費 成功提交位移後 重啟後還是接著消費 這個引數無效

所以3個值的解釋是：

earliset 當各分割槽下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從最早的位移消費

latest 當各分割槽下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分割槽下的資料 none topic各分割槽都存在已提交的offset時，從offset後開始消費；只要有一個分割槽不存在已提交的offset，則丟擲異常

（注意kafka-0.10.1.X版本之前: auto.offset.reset 的值為smallest,和,largest.(offest儲存在zk中) 、

我們這是說的是新版本：kafka-0.10.1.X版本之後: auto.offset.reset 的值更改為:earliest,latest,和none (offest儲存在kafka的一個特殊的topic名為:__consumer_offsets裡面)）

enable.auto.commit 是否自動提交位移

true 自動提交 false需要使用者手動提交 有隻處理一次需要的 最近設定為false自己控制。

fetch.max.bytes consumer單次獲取最大位元組數

max.poll.records 單次poll返回的最大訊息數

預設500條 如果消費很輕量 可以適當提高這個值 增加消費速度。

hearbeat.interval.ms consumer其他組員感知rabalance的時間

該值必須小於 session.timeout.ms 如果檢測到 consumer掛掉 也就根本無法感知rabalance了

connections.max.idle.ms 定期關閉連線的時間

預設是9分鐘 可以設定為-1 永不關閉

更多實時計算，Kafka等相關技術博文，歡迎關注實時流式計算

相關推薦