專為小白打造—Kafka一篇文章入門

Kafka 是MQ消息隊(duì)列作為最常用的中間件之一，其主要特性有：解耦、異步、限流/削峰。

2.1 Topic與Partition

2.2 Broker與Partition

2.3 生產(chǎn)者消費(fèi)者與ZooKeeper

2.4 消費(fèi)者組

消費(fèi)者與消費(fèi)組這種模型可以讓整體的消費(fèi)能力具備橫向伸縮性，我們可以增加(或減少) 消費(fèi)者的個(gè)數(shù)來提高(或降低)整體的消費(fèi)能力。對于分區(qū)數(shù)固定的情況，一味地增加消費(fèi)者 并不會(huì)讓消費(fèi)能力一直得到提升，如果消費(fèi)者過多，出現(xiàn)了消費(fèi)者的個(gè)數(shù)大于分區(qū)個(gè)數(shù)的情況， 就會(huì)有消費(fèi)者分配不到任何分區(qū)。參考下圖（右下），一共有 8 個(gè)消費(fèi)者，7 個(gè)分區(qū)，那么最后的消費(fèi) 者 C7 由于分配不到任何分區(qū)而無法消費(fèi)任何消息。

2.5 ISR、HW、LEO

3.1 注冊信息

3.2 創(chuàng)建主題

3.3 生產(chǎn)者發(fā)送數(shù)據(jù)

3.3.1 封裝ProducerRecord

private final String topic;private final Integer partition;private final Headers headers;private final K key;private final V value;private final Long timestamp;

3.3.2 序列化數(shù)據(jù)、獲取元數(shù)據(jù)、確定分區(qū)

1. 如果ProducerRecord中指定了要發(fā)往那個(gè)分區(qū)，則選擇用戶使用的分區(qū)
2. 如果沒有指定分區(qū)，則查看ProducerRecord中key是否為空，如果不為空則對key進(jìn)行計(jì)算以獲取使用那個(gè)分區(qū)
3. 如果key也為空，則按照輪詢的方式發(fā)送至不同的分區(qū)

3.3.3 寫入緩沖區(qū)、分批分送消息

首先客戶端發(fā)送請求全部會(huì)先發(fā)送給一個(gè)Acceptor，broker里面會(huì)存在3個(gè)線程（默認(rèn)是3個(gè)），這3個(gè)線程都是叫做processor，Acceptor不會(huì)對客戶端的請求做任何的處理，直接封裝成一個(gè)個(gè)socketChannel發(fā)送給這些processor形成一個(gè)隊(duì)列，發(fā)送的方式是輪詢，就是先給第一個(gè)processor發(fā)送，然后再給第二個(gè)，第三個(gè)，然后又回到第一個(gè)。消費(fèi)者線程去消費(fèi)這些socketChannel時(shí)，會(huì)獲取一個(gè)個(gè)request請求，這些request請求中就會(huì)伴隨著數(shù)據(jù)。

線程池里面默認(rèn)有8個(gè)線程，這些線程是用來處理request的，解析請求，如果request是寫請求，就寫到磁盤里。讀的話返回結(jié)果。processor會(huì)從response中讀取響應(yīng)數(shù)據(jù)，然后再返回給客戶端。這就是Kafka的網(wǎng)絡(luò)三層架構(gòu)。

所以如果我們需要對kafka進(jìn)行增強(qiáng)調(diào)優(yōu)，增加processor并增加線程池里面的處理線程，就可以達(dá)到效果。request和response那一塊部分其實(shí)就是起到了一個(gè)緩存的效果，是考慮到processor們生成請求太快，線程數(shù)不夠不能及時(shí)處理的問題。

3.4 消費(fèi)者消費(fèi)數(shù)據(jù)

1. 信息注冊階段，即整個(gè)消費(fèi)者組向集群注冊消費(fèi)信息等
2. 信息消費(fèi)階段，開始信息消息，確保消息可靠性等

3.4.1 信息注冊

3.4.2 消費(fèi)信息

1）拉取消息

   
   

    
    

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

    
    
    
    
try {  while (true) {    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));    for (ConsumerRecord<String, String> record : records) {      int updateCount = 1;      if (map.containsKey(record.value())) {        updateCount = (int) map.get(record.value() + 1);      }      map.put(record.value(), updateCount);    }  }}finally {  consumer.close();}
   
   
   
   

    
   
   

2）反序列化與消費(fèi)消息

3）提交消息位移

• 重復(fù)消費(fèi)。例如從offset為21開始拉取數(shù)據(jù)，拉取到了40，但是當(dāng)消費(fèi)者處理到第30條數(shù)據(jù)的時(shí)候系統(tǒng)宕機(jī)了，那么此時(shí)已經(jīng)提交的offset仍為21，當(dāng)節(jié)點(diǎn)重新連接時(shí)，仍會(huì)從21消費(fèi)，那么此時(shí)21-30的數(shù)據(jù)就會(huì)被重新消費(fèi)。還有一種情況是再均衡時(shí)，例如有新節(jié)點(diǎn)加入也會(huì)引發(fā)類似的問題。
• 消息丟失。

   
   

    
    

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

    
    
    
    
public static void main(String[] args) {    while (true) {        // 這里的參數(shù)指的是輪詢的時(shí)間間隔，也就是多長時(shí)間去拉一次數(shù)據(jù)        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));        records.forEach((ConsumerRecord<String, String> record) -> {            // 模擬消息的處理邏輯            System.out.println("revice: key ===" + record.key() + " value ====" + record.value() + " topic ===" + record.topic());        });        try {            //處理完當(dāng)前批次的消息，在輪詢更多的消息之前，調(diào)用commitSync方法提交當(dāng)前批次最新的消息            consumer.commitSync();        } catch (CommitFailedException e) {            //todo 事務(wù)回滾            e.printStackTrace();        }    }}
   
   
   
   

    
   
   

   
   

    
    

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

    
    
    
    
@Testpublic void asynCommit1(){    while (true) {        // 這里的參數(shù)指的是輪詢的時(shí)間間隔，也就是多長時(shí)間去拉一次數(shù)據(jù)        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));        records.forEach((ConsumerRecord<String, String> record) -> {            System.out.println("revice: key ===" + record.key() + " value ====" + record.value() + " topic ===" + record.topic());        });        consumer.commitAsync();    }}
@Testpublic void asynCommit2(){    while (true) {        // 這里的參數(shù)指的是輪詢的時(shí)間間隔，也就是多長時(shí)間去拉一次數(shù)據(jù)        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));        records.forEach((ConsumerRecord<String, String> record) -> {            System.out.println("revice: key ===" + record.key() + " value ====" + record.value() + " topic ===" + record.topic());        });        // 異步回調(diào)機(jī)制        consumer.commitAsync(new OffsetCommitCallback(){            @Override            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {                if (exception!=null){                    System.out.println(String.format("提交失敗:%s", offsets.toString()));                }            }        });    }}
@Testpublic void asynCommit3(){    while (true) {        // 這里的參數(shù)指的是輪詢的時(shí)間間隔，也就是多長時(shí)間去拉一次數(shù)據(jù)        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));        records.forEach((ConsumerRecord<String, String> record) -> {            System.out.println("revice: key ===" + record.key() + " value ====" + record.value() + " topic ===" + record.topic());        });        consumer.commitAsync((offsets, exception) ->{            if (exception!=null){                System.out.println(String.format("提交失敗:%s", offsets.toString()));            }        });    }}
   
   
   
   

    
   
   

   
   

    
    

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

    
    
    
    
while (true) {    ConsumerRecords records = consumer.poll(100);    for (ConsumerRecord record : records) {        log.trace("Kafka消費(fèi)信息ConsumerRecord={}",record.toString());    }    try {        //先使用異步提交機(jī)制        consumer.commitAsync();    } catch (CommitFailedException e) {        // todo 補(bǔ)償機(jī)制        log.error("commitAsync failed", e)    } finally{      try {        //再使用同步提交機(jī)制        consumer.commitSync();       }  catch (CommitFailedException e) {          // todo 補(bǔ)償機(jī)制          log.error("commitAsync failed", e)      } finally{         consumer.close();       }    }}
   
   

4.1 異常重試

JMQ在消費(fèi)過程中如果有未捕獲的異常會(huì)認(rèn)為消息消費(fèi)失敗，會(huì)首先在本地重試兩次后放入重試隊(duì)列中，進(jìn)入重試隊(duì)列的消息，會(huì)有過期邏輯，當(dāng)超過重試時(shí)間或者超過最大重試次數(shù)后(默認(rèn)3天過期)，消息將會(huì)被丟棄。因此在處理消息時(shí)需要考慮如果出現(xiàn)異常后的處理場景，選擇是重試還是忽略還是記錄數(shù)據(jù)后告警。

   
   

    
    

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

     
     

    
    
    
    
//每隔1min拉取消息ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(60L));for (ConsumerRecord<String, String> record : records) {    try {        //doing    } catch (Exception e) {        //如果此處未捕獲消息，會(huì)直接導(dǎo)致for循環(huán)退出，后續(xù)所有消息都將丟失        log.error("Bdp監(jiān)聽任務(wù)執(zhí)行失敗, taskName:{}", taskName, e);    }}
   
   
   
   

    
   
   

4.2 本地重試與服務(wù)端重試

   
   

    
    

     
     

    
    
    
    
重試分為本地重試和服務(wù)端重試
   
   
   
   

    
   
   

   
   

    
    

     
     

     
     

     
     

     
     

    
    
    
    
<jmq:consumer id="apiConsumer" transport="jmq.apilog.transport">    <!--配置間隔1秒，重試3次-->    <jmq:listener topic="${jmq.topic.apilog}" listener="apiLogMessageListener" retryDelay="1000" maxRetrys="3"/></jmq:consumer>
   
   
   
   

    
   
   

-end-