了解了什么是kafka（ https://www.cnblogs.com/tree1123/p/11226880.html）以后
學習Kafka核心之消費者，kafka的消費者經過幾次版本變化，特別容易混亂，所以一定要搞清楚是哪個版本再研究。
一、舊版本consumer
只有舊版本（0.9以前）才有 high-level consumer 和 low-level consumer之分，很多的文章提到的就是這兩個：低階消費者和高階消費者，低階消費者更靈活但是需要自己維護很多東西，高階就死板一點但是不需要維護太多東西。
high-level consumer就是消費者組。
low-level consumer是單獨一個消費者，單個consumer沒有什么消費者組的概念，與其他consumer相互之間不關聯。
1、low-level consumer
low-level consumer底層實現是
SimpleConsumer 他可以自行管理消費者
Storm的Kafka插件 storm-kafka就是使用了SimpleConsumer
優點是靈活 ， 可以從任意位置拿消息 。
如果需要： 重復讀取數據 只消費部分分區數據 精確消費 就得用這個，
不過必須自己處理位移提交 尋找分區leader broker 處理leader變更。
接口中的方法：
fetch
send 發送請求
getOffsetBefore
commitOffsets
fetchOffsets
earliestOrlatestOffset
close
使用步驟：
參照官網，比較復雜需要好幾步來拉取消息。
Find an active Broker and find out which Broker is the leader for your topic and partition
找到活躍的broker 找到哪個broker是你的topic和partition的leader
Determine who the replica Brokers are for your topic and partition
查出replica 的brokers
Build the request defining what data you are interested in
建立請求
Fetch the data
拿數據
Identify and recover from leader changes
leader變化時恢復
也可以查詢一些offset等metadata信息，具體代碼如下。
//根據指定的分區從主題元數據中找到主副本
SimpleConsumer consumer = new SimpleConsumer(seed, a_port, 100000, 64 * 1024,
"leaderLookup");
List<String> topics = Collections.singletonList(a_topic);
TopicMetadataRequest req = new TopicMetadataRequest(topics); kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);
List<TopicMetadata> metaData = resp.topicsMetadata();
String leader = metaData.leader().host();
//獲取分區的offset等信息
//比如獲取lastoffset
TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
OffsetResponse response = consumer.getOffsetsBefore(request);
long[] offsets = response.offsets(topic, partition);
long lastoffset = offsets[0];
這個api現在應用不多，除非你有特殊需求，比如要自己寫監控，你可能需要更多的元數據信息。
2、high-level consumer
主要使用的類：ConsumerConnector
屏蔽了每個topic的每個Partition的offset的管理（自動讀取zookeeper中該Consumer group的last offset）
Broker失敗轉移，增減Partition Consumer時的負載均衡（當Partiotion和Consumer增減時，Kafka自動負載均衡）
這些功能low-level consumer都需要自己實現的。
主要方法如下：
createMessageStreams
createMessageStreamsByFilter
commitOffsets
setconsumerReblanceListener
shutdown
group通過zookeeper完成核心功能，
zookeeper目錄結構如下：
/consumers/groupId/ids/consumre.id
記錄該consumer的訂閱信息，還被用來監聽consumer存活狀態。這是一個臨時節點，會話失效將會自動刪除。
/consumers/groupId/owners/topic/partition
保存consumer各個消費線程的id，執行rebalance時保存。
/consumers/groupId/offsets/topic/partition
保存該group消費指定分區的位移信息。
這個consumer支持多線程設計，只創建一個consumer實例，但如果是多個分區，將會自動創建多個線程消費。
使用步驟：
Properties properties = new Properties();
properties.put("zookeeper.connect", "ip1:2181,ip2:2181,ip3:2181");//聲明zk
properties.put("group.id", "group03");
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, 1); // 一次從主題中獲取一個數據
Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 獲取每次接收到的這個數據 如果是多線程在這里處理多分區的情況
ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
while(iterator.hasNext()){
String message = new String(iterator.next().message());
System.out.println("接收到: " + message);
}
//auto.offset.reset 默認值為largest
//從頭消費 properties.put("auto.offset.reset", "smallest");
很簡單，我們0.9版本之前使用的很多都是他，集成spring的方法等等。不過0.9版本以后新的consumer出現了。
二、新版本consumer
先說一下版本的問題：
Kafka 0.10.0.0之后 增加了 Kafka Streams 所以Kafka1.0開始Streams 就穩定了。
kafka security 0.9.0.0以后 0.10.0.1之后穩定
0.10.1.0之后 新版本consumer穩定
storm有兩個連kafka的包：
storm-kafka 使用了舊版本的consumer
storm-kafka-client 使用了新版本consumer
kafka 0.9.0.0廢棄了舊版producer和consumer 舊版時scala版 新版用java開發
| 版本 | 推薦producer | 推薦consumer | 原因 |
|---|---|---|---|
| 0.8.2.2 | 舊版 | 舊版 | 新producer尚不穩定 |
| 0.9.0.x | 新版 | 舊版 | 新producer穩定 |
| 0.10.0.x | 新版 | 舊版 | 新consumer不穩定 |
| 0.10.1.0 | 新版 | 新版 | 新consumer穩定 |
| 0.10.2.x | 新版 | 新版 | 都穩定了 |
舊版本中offset管理依托zookeeper，新版本中不在依靠zookeeper。
| 語言 | 包名 | 主要使用類 | |
|---|---|---|---|
| 舊版本 | scala | kafka.consumer.* | ZookeeperConsumerConnector SimpleConsumer |
| 新版本 | java | org.apache.kafka.clients.consumer.* | KafkaConsumer |
新版本的幾個核心概念：
consumer group
消費者使用一個消費者組名（group.id）來標記自己，topic的每條消息都只會發送到每個訂閱他的消費者組的一個消費者實例上。
1、一個消費者組有若干個消費者。
2、對於同一個group，topic的每條消息只能被發送到group下的一個consumer實例上。
3、topic消息可以被發送到多個group中。
consumer端offset
記錄每一個consumer消費的分區的位置
kafka沒有把這個放在服務器端，保存在了consumer group中，並定期持久化。
舊版本會把這個offset定期存在zookeeper中：路徑是 /consumers/groupid/offsets/topic/partitionid
新版本將offset放在了一個內部topic中：__consumer_offsets(前面兩個下划線) 里面有50個分區
所以新版本的consumer就不需要連zookeeper了。
舊版本設置offsets.storage=kafka設置位移提交到這，不常使用。
__consumer_offsets中的結構： key = group.id+topic+partition value=offset
consumer group reblance
單個consumer是沒有rebalance的。
他規定了一個consumer group下的所有consumer如何去分配所有的分區。
單線程示例代碼：
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）。
Properties詳解：
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 永不關閉
poll方法詳解：
（舊版本：多分區多線程 新版本：一個線程管理多個socket連接）
但新版本KafkaConsumer是雙線程的，主線程負責：消息獲取，rebalance，coordinator，位移提交等等，
另一個是后台心跳線程。
根據上邊的各種配置，poll方法會找到offset，當獲取了足夠多的可用數據，或者等待時間超過了指定的超時時間，就會返回。
java consumer不是線程安全的，同一個KafkaConsumer用在了多個線程中，將會報Kafka Consumer is not safe for multi-threaded assess異常。可以加一個同步鎖進行保護。
poll的超時參數，已經說過1000的話是超時設定，如果沒有很多數據，也就等一秒，就返回了，比如定時5秒的將消息寫入，就可以將超時參數設置為5000，達到效率最大化。
如果沒有定時任務呢，那就設置為 Long.MAX_VALUE 未獲取足夠多的數據就無限等待。這里要捕獲一下WakeupException。
consumer offset詳解：
consumer需要定期向kafka提交自己的offset信息。已經學過 新版本將他提交到了一個topic中 __consumer_offsets。
offset有一個更大的作用是實現交付語義：
最多一次 at most once 可能丟失 不會重復
最少一次 at least once 可能重復 不會丟失
精確一次 exactly once 不丟失 不重復 就一次
若consumer在消費之前提交位移 就實現了at most once
若是消費后提交 就實現了 at least once 默認是這個。
consumer的多個位置信息：
上次提交的位置 當前位置 水位 日志最新位移
0 1 。。 5 。。 10 。。 15
上次提交位置：consumer最近一次提交的offset值;
當前位置：consumer上次poll 到了這個位置 但是還沒提交;
水位：這是分區日志的管理 consumer無法讀取水位以上的消息;
最新位移： 也是分區日志的管理 最大的位移值 一定不會比水位小。
新版本的consumer會在broker選一個broker作為consumergroup的coordinator，用於實現組成員管理，消費分配方案，提交位移。如果consumer崩潰，他負責的分區就分配給其他consumer，如果沒有做好位移提交就可能重復消費。
多次提交的情況，kafka只關注最新一次的提交。
默認consumer自動提交位移 提交間隔為5秒 可以通過 auto.commit.interval.ms 設置這個間隔。
自動提交可以減少開發，但是可能重復消費，所以需要精准消費時還是要手動提交。設置手動提交 enable.auto.commit = false，然后調用 consumer.commitSync() 或者 consumer.commitAync() Sync為同步方式，阻塞 Aync為異步方式，不會阻塞。這兩個方法可以傳參，指定為哪個分區提交，這樣更合理一些。
（舊版本的自動提交設置是 auto.commit.enable 默認間隔為60秒）
rebalance詳解：
rebalance是consumer group如何分配topic的所有分區。
正常情況，比如有10個分區，5個consumer 那么consumer group將為每個consumer 平均分配兩個分區。
每個分區只會分給一個consumer實例。有consumer出現問題，會重新執行這個過程，這個過程就是rebalance。
（舊版本通過zookeeper管理rebalance，新版本會選取某個broker為group coordinator來管理）
rebalance的觸發條件：
1、有新的consumer加入，或者有consumer離開或者掛掉。
2、group訂閱的topic發生變更，比如正則訂閱。
3、group訂閱的分區數發生變化。
第一個經常出現，不一定是掛掉，也可能是處理太慢，為了避免頻繁rebalance，要調整好request.timeout.ms max.poll.records和ma.poll.interval.
rebalance分區策略：
partition.assignment.strategy 設置 自定義分區策略-創建分區器 assignor
range策略（默認），將分區划分為分區段，一次分配給每個consumer。
round-robin策略，輪詢分配。
sticky策略（0.11.0.0出現，更優秀），range策略在訂閱多個topic時會不均勻。
sticky有兩個原則，當兩者發生沖突時，第一個目標優先於第二個目標。
- 分區的分配要盡可能的均勻；
- 分區的分配盡可能的與上次分配的保持相同。
rebalance generation分代機制保證rabalance時重復提交的問題，延遲的offset提交時舊的generation信息會報異常ILLEGAL_GENERATION
rebalance過程：
1、確定coordinator所在的broker，建立socket連接。
確定算法： Math.abs(groupID.hashCode) % offsets.topic.num.partition 參數值（默認50）
尋找__consumer_offset分區50的leader副本所在的broker，該broker即為這個group的coordinator
2、加入組
所有consumer會向coordinator發送JoinGroup請求，收到所有請求后選一個consumer做leader（這個leader是consumer coordinator是broker），coordinator把成員和訂閱信息發給coordinator。
3、同步分配方案
leader制定分配方案，通過SyncGroup請求發給coordinator，每個consumer也會發請求返回方案。
kafka也支持offset不提交到__consumer_offset，可以自定義，這時候就需要實現一個監聽器ConsumerRebalanceListener，在這里重新處理Rebalance的邏輯。
多線程示例代碼：
這里要根據自身需求開發，我這里只舉一個簡單的例子，就是幾個分區就啟動幾個consumer，一一對應。
三個類：
Main：
public static void main(String[] args) {
String bootstrapServers = "kafka01:9092，kafka02:9092";
String groupId = "test";
String topic = "testtopic";
int consumerNum = 3;
ConsumerGroup cg = new ConsumerGroup(consumerNum,bootstrapServers,groupId,topic);
cg.execute();
}
import java.util.ArrayList;
import java.util.List;
public class ConsumerGroup {
private List<ConsumerRunnable> consumers;
public ConsumerGroup(int consumerNum,String bootstrapServers,String groupId,String topic){
consumers = new ArrayList<>(consumerNum);
for(int i=0;i < consumerNum;i++){
ConsumerRunnable ConsumerRunnable = new ConsumerRunnable(bootstrapServers,groupId,topic);
consumers.add(ConsumerRunnable);
}
}
public void execute(){
for(ConsumerRunnable consumerRunnable:consumers){
new Thread(consumerRunnable).start();
}
}
}
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 ConsumerRunnable implements Runnable{
private final KafkaConsumer<String,String> consumer;
public ConsumerRunnable(String bootstrapServers,String groupId,String topic){
Properties props = new Properties();
props.put("bootstrap.servers", bootstrapServers);
props.put("group.id", groupId);
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");
this.consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic));
}
@Override
public void run() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(10);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}
}
}
standalone consumer
有一些需求，需要指定一個消費者消費某一個分區。彼此之間不干擾，一個standalone consumer崩潰不會影響其他。
類似舊版本的低階消費者。
示例代碼如下：consumer.assign方法訂閱分區
public static void main(String[] args) {
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);
List<TopicPartition> partitions = new ArrayList<>();
List<PartitionInfo> allpartitions = consumer.partitionsFor("testtopic");
if(allpartitions!=null && !allpartitions.isEmpty()){
for(PartitionInfo partitionInfo:allpartitions){
partitions.add(new TopicPartition(partitionInfo.topic(),partitionInfo.partition()));
}
consumer.assign(partitions);
}
while (true) {
ConsumerRecords<String, String> records = consumer.poll(10);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}
}
以上為kafka消費者的學習，不同的具體細節還需要通過官網文檔仔細學習。
更多實時計算,Flink,Kafka等相關技術博文，歡迎關注實時流式計算
