一、Kafka 概述

Kafka 是一個分布式的基于發布/訂閱模式的訊息佇列（Message Queue），主要應用于大資料實時處理領域，

訊息佇列的兩種模式

點對點模式

一對一，消費者主動拉取資料，訊息收到后訊息清除，

訊息生產者生產訊息發送到Queue中，然后訊息消費者從Queue中取出并且消費訊息，訊息被消費以后，queue 中不再存盤，Queue 支持存在多個消費者，但是對一個訊息而言，只會有一個消費者可以消費，

發布/訂閱模式

一對多，消費者消費資料之后不會清除消，
訊息生產者（發布）將訊息發布到 topic 中，同時有多個訊息消費者（訂閱）消費該訊息，和點對點方式不同，發布到 topic 的訊息會被所有訂閱者消費，kafka就是基于這種模式

Kafka 基礎架構

1）Producer ：訊息生產者，就是向 kafka broker 發訊息的客戶端；
2）Consumer ：訊息消費者，向 kafka broker 取訊息的客戶端；
3）Consumer Group （CG）：消費者組，由多個 consumer 組成，消費者組內每個消費者負責消費不同磁區的資料，一個磁區只能由一個組內消費者消費；消費者組之間互不影響，所有的消費者都屬于某個消費者組，即消費者組是邏輯上的一個訂閱者，
4）Broker ：一臺 kafka 服務器就是一個 broker，一個集群由多個 broker 組成，一個 broker可以容納多個 topic，
5）Topic ：可以理解為一個佇列，生產者和消費者面向的都是一個 topic；
6）Partition：為了實作擴展性，一個非常大的 topic 可以分布到多個 broker（即服務器）上，一個 topic 可以分為多個 partition，每個 partition 是一個有序的佇列；
7）Replica：副本，為保證集群中的某個節點發生故障時，該節點上的 partition 資料不丟失，且 kafka 仍然能夠繼續作業，kafka 提供了副本機制，一個 topic 的每個磁區都有若干個副本，一個 leader 和若干個 follower，
8）leader：每個磁區多個副本的“主”，生產者發送資料的物件，以及消費者消費資料的物件都是 leader，
9）follower：每個磁區多個副本中的“從”，實時從 leader 中同步資料，保持和 leader 資料的同步，leader 發生故障時，某個 follower 會成為新的 follower，

二、docker 安裝 kafka

輸入下面兩條命令即可快速安裝單機zookeeper和kafka

docker run -d --name zookeeper -p 2181:2181  wurstmeister/zookeeper
docker run -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 --link zookeeper -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.169.129(機器的ip):9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

通過kafka自帶工具生產消費訊息測驗:
1、首先,進入到kafka的docker容器中

docker exec -it kafka sh

2、運行消費者,進行訊息的監聽

kafka-console-consumer.sh --bootstrap-server 192.168.169.129:9092 --topic kafeidou --from-beginning

3、打開一個新的ssh視窗,同樣進入kafka的容器中,執行下面這條命令生產訊息

kafka-console-producer.sh --broker-list 192.168.169.129:9092 --topic kafeidou

輸入完這條命令后會進入到控制臺，可以輸入任何想發送的訊息,這里發送一個hello

>>
>hello
>

可以看到,在生產者的控制臺中輸入訊息后,消費者的控制臺立刻看到了訊息,到目前為止,一個kafka完整的hello world就完成了.kafka的部署加上生產者消費者測驗，

另一種方式部署
通過docker-compose部署kafka，
首先創建一個docker-compose.yml檔案

version: '3.7'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    volumes:
      - ./data:/data # 掛載目錄，存盤zookeeper資料
    ports:
      - 2182:2181
       
  kafka9094:
    image: wurstmeister/kafka
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 0
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.60:9092
      KAFKA_CREATE_TOPICS: "kafeidou:2:0"   #kafka啟動后初始化一個有2個partition(磁區)0個副本名叫kafeidou的topic 
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
    volumes:
      - ./kafka-logs:/kafka # 掛載目錄，存盤kafka資料
    depends_on:
      - zookeeper

部署起來很簡單,在docker-compose.yml檔案的目錄下執行docker-compose up -d就可以了,測驗方式和上面的一樣，
這個docker-compose做的東西比上面docker方式部署的東西要多一些

• 資料持久化，在當前目錄下掛載了兩個目錄分別存盤zookeeper和kafka的資料,當然在docker run 命令中添加 -v
  選項也是可以做到這樣的效果的
• kafka在啟動后會初始化一個有磁區的topic,同樣的,docker run的時候添加 -e KAFKA_CREATE_TOPICS=kafeidou:2:0 也是可以做到的，

優先推薦docker-compose方式部署：
因為單純使用docker方式部署的話，如果有改動(例如:修改對外開放的埠號)的情況下,docker需要把容器停止docker stop 容器ID/容器NAME,然后洗掉容器docker rm 容器ID/容器NAME,最后啟動新效果的容器docker run …
而如果在docker-compose部署的情況下如果修改內容只需要修改docker-compose.yml檔案對應的地方,例如2181:2181改成2182:2182,然后再次在docker-compose.yml檔案對應的目錄下執行docker-compose up -d就能達到更新后的效果，

三、Kafka 架構深入

Kafka 作業流程及檔案存盤機制

Kafka 作業流程：

1、Kafka 中訊息是以 topic 進行分類的，生產者生產訊息，消費者消費訊息，都是面向 topic的，
2、topic 是邏輯上的概念，而partition是物理上的概念，每個partition 對應于一個 log 檔案，該log檔案中存盤的就是producer生產的資料
3、Producer 生產的資料會被不斷追加到該log 檔案末端，且每條資料都有自己的 offset，
4、消費者組中的每個消費者，都會實時記錄自己消費到了哪個 offset，以便出錯恢復時，從上次的位置繼續消費，

Kafka檔案存盤機制：

1、由于生產者生產的訊息會不斷追加到 log 檔案末尾，為防止 log 檔案過大導致資料定位效率低下，Kafka 采取了分片和索引機制，將每個 partition 分為多個 segment，每個 segment對應兩個檔案——“.index”檔案和“.log”檔案，
2、這些檔案位于一個檔案夾下，該檔案夾的命名規則為：topic 名稱+磁區序號，例如，first 這個 topic 有三個磁區，則其對應的檔案夾為 first-0,first-1,first-2，
例：

index 和 log 檔案以當前 segment 的第一條訊息的 offset 命名，
下圖為 index 檔案和 log檔案的結構示意圖，

“.index”檔案存盤大量的索引資訊，“.log”檔案存盤大量的資料，索引檔案中的元資料指向對應資料檔案中 message 的物理偏移地址，

Kafka 生產者

磁區策略

磁區的原因：
（1）方便在集群中擴展，每個 Partition 可以通過調整以適應它所在的機器，而一個 topic又可以有多個 Partition 組成，因此整個集群就可以適應任意大小的資料了；
（2）可以提高并發，因為可以以 Partition 為單位讀寫了

磁區的原則：
生產者發送的訊息，發到哪個磁區呢？
我們需要將 producer 發送的資料封裝成一個 ProducerRecord 物件，

（1）指明 partition 的情況下，直接將指明的值直接作為 partiton 值；
（2）沒有指明 partition 值但有 key 的情況下，將 key 的 hash 值與 topic 的 partition數進行取余得到 partition 值；
（3）既沒有 partition 值又沒有 key 值的情況下，第一次呼叫時隨機生成一個整數（后面每次呼叫在這個整數上自增），將這個值與 topic 可用的 partition 總數取余得到 partition 值，也就是常說的 round-robin 演算法，

生產者發來的訊息通過這種方式，就這樣分散存盤到不同的磁區中，

資料可靠性保證

為保證 producer 發送的資料，能可靠的發送到指定的 topic，topic 的每個 partition 收到producer 發送的資料后，都需要向 producer 發送 ack（acknowledgement 確認收到），如果producer 收到 ack，就會進行下一輪的發送，否則重新發送資料，

副本資料同步策略

Kafka 選擇了第二種方案，原因如下：
1.同樣為了容忍 n 臺節點的故障，第一種方案需要 2n+1 個副本，而第二種方案只需要 n+1個副本，而 Kafka 的每個磁區都有大量的資料，第一種方案會造成大量資料的冗余，
2.雖然第二種方案的網路延遲會比較高，但網路延遲對 Kafka 的影響較小，

ISR

采用第二種方案之后，設想以下情景：leader 收到資料，所有 follower 都開始同步資料，但有一個 follower，因為某種故障，遲遲不能與 leader 進行同步，那 leader 就要一直等下去，直到它完成同步，才能發送 ack，這個問題怎么解決呢？
Leader 維護了一個動態的 in-sync replica set (ISR)，意為和 leader 保持同步的 follower 集合，當 ISR 中的 follower 完成資料的同步之后，leader 就會給 follower 發送 ack，如果 follower長時間 未 向 leader 同 步 數 據 ， 則 該 follower 將 被 踢 出 ISR ， 該 時 間 閾 值 由replica.lag.time.max.ms 引數設定，Leader 發生故障之后，就會從 ISR 中選舉新的 leader，

ack應答機制

對于某些不太重要的資料，對資料的可靠性要求不是很高，能夠容忍資料的少量丟失，所以沒必要等 ISR 中的 follower 全部接收成功，
Kafka 為用戶提供了三種可靠性級別，用戶根據對可靠性和延遲的要求進行權衡，選擇以下的配置，
acks 引數配置
acks：

• 0：producer 不等待 broker 的 ack，這一操作提供了一個最低的延遲，broker 一接收到還沒有寫入磁盤就已經回傳，當broker 故障時有可能丟失資料；
• 1：producer 等待 broker 的 ack，partition 的 leader 落盤成功后回傳 ack，如果在follower同步成功之前 leader 故障，那么將會丟失資料；
• -1（all）：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盤成功后才回傳 ack，但是如果在 follower 同步完成后，broker 發送 ack 之前，leader發生故障，那么會造成資料重復，

面試大廠時，一旦簡歷上寫了Kafka，幾乎必然會被問到一個問題：說說acks引數對訊息持久化的影響？

資料一致性問題，故障處理細節

LEO：指的是每個副本最大的 offset；
HW：指的是消費者能見到的最大的 offset，ISR 佇列中最小的 LEO，
（1）follower 故障
follower 發生故障后會被臨時踢出 ISR，待該 follower 恢復后，follower 會讀取本地磁盤記錄的上次的 HW，并將 log 檔案高于 HW 的部分截取掉，從 HW 開始向 leader 進行同步，等該 follower 的 LEO 大于等于該 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加ISR 了，
（2）leader 故障
leader 發生故障之后，會從 ISR 中選出一個新的 leader，之后，為保證多個副本之間的資料一致性，其余的 follower 會先將各自的 log 檔案高于 HW 的部分截掉，然后從新的 leader同步資料，
注意：這只能保證副本之間的資料一致性，并不能保證資料不丟失或者不重復，

Exactly Once 語意

服務器的 ACK 級別設定為-1，可以保證 Producer 到 Server 之間不會丟失資料，但是不能保證資料不重復，即 At Least Once 語意，
將服務器 ACK 級別設定為 0，可以保證生產者每條訊息只會被發送一次，但是不能保證資料不丟失，即 At Most Once 語意，
對于一些非常重要的資訊，比如說交易資料，下游資料消費者要求資料既不重復也不丟失，即 Exactly Once 語意，

0.11 版本的 Kafka，引入了一項重大特性：冪等性，所謂的冪等性就是指 Producer 不論向 Server 發送多少次重復資料，Server 端都只會持久化一條，
即：At Least Once + 冪等性 = Exactly Once

要啟用冪等性，只需要將 Producer 的引數中 enable.idompotence 設定為 true 即可，

Kafka的冪等性實作其實就是將原來下游需要做的去重放在了資料上游，開啟冪等性的 Producer 在初始化的時候會被分配一個 PID，發往同一 Partition 的訊息會附帶 Sequence Number，而Broker 端會對<PID, Partition, SeqNumber>做快取，當具有相同主鍵的訊息提交時，Broker 只會持久化一條，
但是 PID 重啟就會變化，同時不同的 Partition 也具有不同主鍵，所以冪等性無法保證跨磁區跨會話的 Exactly Once，

Kafka 消費者

消費方式

consumer 采用 pull（拉）模式從 broker 中讀取資料，
push（推）模式很難適應消費速率不同的消費者，因為訊息發送速率是由 broker 決定的，它的目標是盡可能以最快速度傳遞訊息，但是這樣很容易造成 consumer 來不及處理訊息，典型的表現就是拒絕服務以及網路擁塞，而 pull 模式則可以根據 consumer 的消費能力以適
當的速率消費訊息，

pull 模式不足之處是，如果 kafka 沒有資料，消費者可能會陷入回圈中，一直回傳空資料，針對這一點，Kafka 的消費者在消費資料時會傳入一個時長引數 timeout，如果當前沒有資料可供消費，consumer 會等待一段時間之后再回傳，這段時長即為 timeout，

磁區分配策略

一個 consumer group 中有多個 consumer，一個 topic 有多個 partition，所以必然會涉及到 partition 的分配問題，即確定那個 partition 由哪個 consumer 來消費，

Kafka 有兩種分配策略，一是 RoundRobin，一是 Range，

RoundRobin(輪詢)

使用的前提是消費組里的消費者訂閱的主題都是一樣的，面向組

按順序分配磁區給消費者

輪詢的方式分配磁區時，不是基于同一個topic去分配的，而是把所有磁區看成一個整體，TopicAndPartition類就代表著所有topic的磁區，TopicAndPartition會根據磁區物件的hash值排序，然后把排完序的磁區輪詢分配給消費者，

這種方式會存在什么問題呢？
如果消費組里的消費者A, 只訂閱了topic甲和乙，消費者B訂閱的是乙和丙，但是通過輪序后，消費者A就能分配到丙的磁區，消費者B就能分配到甲的磁區，這就不好了，所以使用輪詢的方式，前提是消費組里的消費者都是訂閱的相同的主題，這就沒問題了，

Range（范圍）

默認的消費者磁區分配方式，基于同一個主題，磁區數 / 消費組里的消費者數，然后均分下去，如果除不均，會有一個訊息者多分配磁區，面向topic

如果有很多topic, 都除不均， 那么每個topic里多的磁區都會分配給同一個消費者，這個消費者相比其他消費者，會更有壓力

什么時候觸發磁區分配策略

當消費者組里的消費者個數發生變化時，

offset 的維護

由于 consumer 在消費程序中可能會出現斷電宕機等故障，consumer 恢復后，需要從故障前的位置的繼續消費，所以 consumer 需要實時記錄自己消費到了哪個 offset，以便故障恢復后繼續消費，

從 0.9 版本開始，consumer 默認將 offset 保存在 Kafka 一個內置的 topic 中，該 topic 為__consumer_offsets，

讀取__consumer_offsets里的offset：
1）修改組態檔 consumer.properties

exclude.internal.topics=false
Kafka 中有兩個內部的主題： __consumer_offsets 和 __transaction_state，
exclude.internal.topics 用來指定 Kafka 中的內部主題是否可以向消費者公開，默認值為 true，
如果設定為 true，那么只能使用 subscribe(Collection)的方式而不能使用 subscribe(Pattern)的方式來訂閱內部主題，設定為 false 則沒有這個限制，

2）讀取 offset
0.11.0.0 之前版本:

bin/kafka-console-consumer.sh --topic __consumer_offsets --
zookeeper hadoop102:2181 --formatter 
"kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter" 
--consumer.config config/consumer.properties --from-beginning

0.11.0.0 之后版本(含):

bin/kafka-console-consumer.sh --topic __consumer_offsets --
zookeeper hadoop102:2181 --formatter 
"kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageForm
atter" --consumer.config config/consumer.properties --frombeginning

Kafka 高效讀寫資料

順序寫磁盤
Kafka 的 producer 生產資料，要寫入到 log 檔案中，寫的程序是一直追加到檔案末端，為順序寫，官網有資料表明，同樣的磁盤，順序寫能到 600M/s，而隨機寫只有 100K/s，這與磁盤的機械機構有關，順序寫之所以快，是因為其省去了大量磁頭尋址的時間，

零復制技術

Zookeeper 在 Kafka 中的作用

Kafka 集群中有一個 broker 會被選舉為 Controller，負責管理集群 broker 的上下線，所有 topic 的磁區副本分配和 leader 選舉等作業，
Controller 的管理作業都是依賴于 Zookeeper 的，

以下為 partition 的 leader 選舉程序：

Kafka 事務

Kafka 從 0.11 版本開始引入了事務支持，事務可以保證 Kafka 在 Exactly Once 語意的基礎上，生產和消費可以跨磁區和會話，要么全部成功，要么全部失敗，

Producer 事務：
為了實作跨磁區跨會話的事務，需要引入一個全域唯一的 Transaction ID，并將 Producer獲得的PID 和Transaction ID 系結，這樣當Producer 重啟后就可以通過正在進行的 TransactionID 獲得原來的 PID，

為了管理 Transaction，Kafka 引入了一個新的組件 Transaction Coordinator，Producer 就是通過和 Transaction Coordinator 互動獲得 Transaction ID 對應的任務狀態，Transaction Coordinator 還負責將事務所有寫入 Kafka 的一個內部 Topic，這樣即使整個服務重啟，由于事務狀態得到保存，進行中的事務狀態可以得到恢復，從而繼續進行，

Consumer 事務：
上述事務機制主要是從 Producer 方面考慮，對于 Consumer 而言，事務的保證就會相對較弱，尤其時無法保證 Commit 的資訊被精確消費，這是由于 Consumer 可以通過 offset 訪問任意資訊，而且不同的 Segment File 生命周期不同，同一事務的訊息可能會出現重啟后被洗掉的情況，

四、API

Springboot 整合

        <!--引入kafka依賴-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.4.1.RELEASE</version>
        </dependency>
spring:
  kafka:
    # kafka集群資訊 127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094
    bootstrap-servers: 192.168.169.129：9092
    # 生產者配置
    producer:
      # 設定大于0的值，則客戶端會將發送失敗的記錄重新發送
      retries: 3
      # 批次大小，16K，只有資料積累到 batch.size 之后，sender 才會發送資料，
      batch-size: 16384
      properties:
        # 等待時間，1ms，如果資料遲遲未達到 batch.size，sender 等待 linger.time 之后就會發送資料，
        linger.ms: 1
      #32M，RecordAccumulator緩沖區大小
      buffer-memory: 33554432
      # ack應答級別
      acks: 1
      # 指定訊息key和訊息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      # 消費者組Id
      group-id: zhTestGroup
      # 關閉自動提交
      enable-auto-commit: false
      # 當各磁區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 當每一條記錄被消費者監聽器（ListenerConsumer）處理之后提交
      # RECORD
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之后提交
      # BATCH
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之后，距離上次提交時間大于TIME時提交
      # TIME
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之后，被處理record數量大于等于COUNT時提交
      # COUNT
      # TIME |　COUNT　有一個條件滿足時提交
      # COUNT_TIME
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之后, 手動呼叫Acknowledgment.acknowledge()后提交
      # MANUAL
      # 手動呼叫Acknowledgment.acknowledge()后立即提交，一般使用這種
      # MANUAL_IMMEDIATE
      ack-mode: manual_immediate



  kafka:
    # bootstrap-servers: ${KAFKA_URL:121.89.205.245:9002} 
    bootstrap-servers: ${KAFKA_HOST:121.89.205.245}:${KAFKA_PORT:9092}
    producer:
      batch-size: 16
      retries: 0
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        security.protocol: SASL_PLAINTEXT
        sasl.mechanism: PLAIN
        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username=${KAFKA_USERNAME:enduser} password=${KAFKA_PASSWORD:u769q83g};
        acks: 0
    consumer:
      group-id: ${KAFKA_CONSUMER_GROUP_ID:devicesGrop-dev-lzh}
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        security.protocol: SASL_PLAINTEXT
        sasl.mechanism: PLAIN
        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username=${KAFKA_USERNAME:enduser} password=${KAFKA_PASSWORD:u769q83g};
        acks: 0         

1、簡單的發訊息和消費訊息：

@Autowired
private KafkaTemplate kafkaTemplate;
 @RequestMapping("/send")
 public void send() {
     kafkaTemplate.send("lzhTest01", "test");
 }
 @Component
public class MyKafkaConsumer {
    @KafkaListener(topics = "lzhTest01", groupId = "zhTestGroup")
    public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手動提交offset
        ack.acknowledge();
    }
}

2、帶回呼的生產者

/**
 * 帶回呼的生產者
 */
@RequestMapping("/callback/send")
public void callbackSend()  {
    for (int i = 0; i < 10; i++) {
    	// 第二個引數，指定訊息發到磁區1中
        kafkaTemplate.send(TOPIC_NAME,1,"lzh","test" +i).addCallback(new ListenableFutureCallback<SendResult<String, Object>>(){
            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                RecordMetadata recordMetadata = stringObjectSendResult.getRecordMetadata();
                System.out.println(recordMetadata.partition() + " " + recordMetadata.offset());
            }
            @Override
            public void onFailure(Throwable throwable) {
                throwable.printStackTrace();
            }
        });
    }
}

3、自定義磁區器
發訊息不指定磁區的話，使用默認的磁區器

public class MyPartitioner implements Partitioner {
    /**
     * 當前訊息到topic時，沒有指定磁區，就通過這個方法分配一個磁區
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        return 1;
    }
    @Override
    public void close() {

    }
    @Override
    public void configure(Map<String, ?> map) {
    }
}

同時組態檔指定磁區器

4、同步發送生產者

kafkaTemplate.send(TOPIC_NAME, "test" +i).get();

5、批量消費訊息
方式一：直接在組態檔中配置，這種方式，消費者的入參都必須是List<ConsumerRecord<?, ?>> list

@Component
public class MyKafkaConsumer {
    @KafkaListener(topics = "lzhTest01", groupId = "zhTestGroup")
    public void listenZhugeGroup(List<ConsumerRecord<?, ?>> list) {
        for (ConsumerRecord<?, ?> consumerRecord : list) {
            System.out.println(consumerRecord.key() + "  " + consumerRecord.value());
        }
    }
    @KafkaListener(topics = "lzhTest02", groupId = "zhTestGroup")
    public void listenZhugeGroup(String message, Acknowledgment ack) {
        System.out.println(message);
        ack.acknowledge();
    }
}

方式二：可能有些消費者就想一條一條的消費，有些消費者想批量消費

@Configuration
public class KafkaConfig {
    @Autowired
    KafkaAdmin kafkaAdmin;
    @Autowired
    KafkaProperties properties;
    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin.getConfig());
    }
    /**
     * 批量消費配置
     * @return
     */
    @Bean
    public KafkaListenerContainerFactory<?> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(this.properties.buildConsumerProperties()));
        // 開啟批量監聽
        factory.setBatchListener(true);
        //設定提交偏移量的方式
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

6、重置offset

earliest
當各磁區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費
latest
當各磁區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該磁區下的資料
none
topic各磁區都存在已提交的offset時，從offset后開始消費；只要有一個磁區不存在已提交的offset，則拋出例外

# 消費者換了組 或 offset失效了，該配置生效
# earliest當各磁區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費
auto-offset-reset: earliest

7、手動提交offset

只有設定了手動提交，且設定了ack-mode，消費者入參才能添加Acknowledgment ack，才能ack.acknowledge();手動提交事務

    @KafkaListener(topics = "lzhTest01", groupId = "zhTestGroup")
    public void listenZhugeGroup(List<ConsumerRecord<?, ?>> list, Acknowledgment ack) {
        for (ConsumerRecord<?, ?> consumerRecord : list) {
            System.out.println(consumerRecord.key() + "  " + consumerRecord.value());
        }
        // 如果不提交，就會出現訊息消費了，但是offset還是在之前的位置
        // ack.acknowledge()后提交，如果消費了訊息，還沒提交，kafka掛了，因為沒提交，offset沒移動，會重復消費訊息
        // ack.acknowledge()先提交，再消費，如果提交了，然后消費失敗了，就丟訊息  
        ack.acknowledge();
    }

8、操作topic
注入AdminClient

@Configuration
public class KafkaConfig {
    @Autowired
    KafkaAdmin kafkaAdmin;
    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin.getConfig());
    }
}

topic工具類：

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.config.ConfigResource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.concurrent.ExecutionException;
/**
 * 主要功能包括：
 *
 * 創建Topic：createTopics(Collection<NewTopic> newTopics)
 * 洗掉Topic：deleteTopics(Collection<String> topics)
 * 顯示所有Topic：listTopics()
 * 查詢Topic：describeTopics(Collection<String> topicNames)
 * 查詢集群資訊：describeCluster()
 * 查詢ACL資訊：describeAcls(AclBindingFilter filter)
 * 創建ACL資訊：createAcls(Collection<AclBinding> acls)
 * 洗掉ACL資訊：deleteAcls(Collection<AclBindingFilter> filters)
 * 查詢配置資訊：describeConfigs(Collection<ConfigResource> resources)
 * 修改配置資訊：alterConfigs(Map<ConfigResource, Config> configs)
 * 修改副本的日志目錄：alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment)
 * 查詢節點的日志目錄資訊：describeLogDirs(Collection<Integer> brokers)
 * 查詢副本的日志目錄資訊：describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas)
 * 增加磁區：createPartitions(Map<String, NewPartitions> newPartitions)
 */
@Component
public class KafkaAdminUtil {
    @Autowired
    private AdminClient adminClient;
    /**
     * 創建主題
     */
    public CreateTopicsResult createTopic(String topic, int numPartitions) throws InterruptedException {
        NewTopic newTopic = new NewTopic(topic, numPartitions, (short) 1);
        return adminClient.createTopics(Arrays.asList(newTopic));
    }

    /**
     * 查看主題
     */
    public void selectTopicInfo(String topic) throws ExecutionException, InterruptedException {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList(topic));
        KafkaFuture<Map<String, TopicDescription>> all = result.all();
        Map<String, TopicDescription> stringTopicDescriptionMap = all.get();
        Set<String> set = stringTopicDescriptionMap.keySet();
        for (String k : set) {
            TopicDescription topicDescription = stringTopicDescriptionMap.get(k);
            System.out.println("k: " + k + " ,v: " + topicDescription.toString() + "\n");
        }
    }

    /**
     * 洗掉主題
     */
    public DeleteTopicsResult deleteTopic(String topic){
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topic));
        return deleteTopicsResult;
    }

    /**
     * topic配置描述
     */
    public void describeConfig() throws ExecutionException, InterruptedException {
        DescribeConfigsResult ret = adminClient.describeConfigs(Collections.singleton(new ConfigResource(ConfigResource.Type.TOPIC, "batch_topic")));
        Map<ConfigResource, Config> configs = ret.all().get();
        for (Map.Entry<ConfigResource, Config> entry : configs.entrySet()) {
            ConfigResource key = entry.getKey();
            Config value = entry.getValue();
            System.out.println(String.format("Resource type: %s, resource name: %s", key.type(), key.name()));
            Collection<ConfigEntry> configEntries = value.entries();
            for (ConfigEntry each : configEntries) {
                System.out.println(each.name() + " = " + each.value());
            }
        }
    }

    /**
     * 集群描述
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void describeCluster() throws ExecutionException, InterruptedException {
        DescribeClusterResult ret = adminClient.describeCluster();
        System.out.println(String.format("Cluster id: %s, controller: %s", ret.clusterId().get(), ret.controller().get()));
        System.out.println("Current cluster nodes info: ");
        for (Node node : ret.nodes().get()) {
            System.out.println(node);
        }

        Thread.sleep(1000);
    }

    /**
     * 更新topic配置
     */
    public void alterConfigs() throws ExecutionException, InterruptedException {
        Config topicConfig = new Config(Arrays.asList(new ConfigEntry("cleanup.policy", "compact")));
        adminClient.alterConfigs(Collections.singletonMap(
                new ConfigResource(ConfigResource.Type.TOPIC, "batch_topic"), topicConfig)).all().get();
    }

    /**
     * 洗掉指定主題
     */
    public void deleteTopics() throws ExecutionException, InterruptedException {
        KafkaFuture<Void> futures = adminClient.deleteTopics(Arrays.asList("batch_topic")).all();
        futures.get();
    }

    /**
     * 描述給定的主題
     */
    public void describeTopics() throws ExecutionException, InterruptedException {
        DescribeTopicsResult ret = adminClient.describeTopics(Arrays.asList("batch_topic", "__consumer_offsets"));
        Map<String, TopicDescription> topics = ret.all().get();
        for (Map.Entry<String, TopicDescription> entry : topics.entrySet()) {
            System.out.println(entry.getKey() + " ===> " + entry.getValue());
        }
    }

    /**
     * 列印集群中的所有主題
     */
    public void listAllTopics() throws ExecutionException, InterruptedException {
        ListTopicsOptions options = new ListTopicsOptions();
        // 包括內部主題，如_consumer_offsets
        options.listInternal(true);
        ListTopicsResult topics = adminClient.listTopics(options);
        Set<String> topicNames = topics.names().get();
        System.out.println("Current topics in this cluster: " + topicNames);
    }
}

/**
 * topic測驗
 */
@RequestMapping("/topic/send")
public void topic() throws ExecutionException, InterruptedException {
//        DeleteTopicsResult deleteTopicsResult = kafkaAdminUtil.deleteTopic(TOPIC_NAME);
    CreateTopicsResult topic = kafkaAdminUtil.createTopic(TOPIC_NAME, 2);
    Thread.sleep(100);
    kafkaAdminUtil.selectTopicInfo(TOPIC_NAME);
}

Producer API

訊息發送流程

Kafka 的 Producer 發送訊息采用的是異步發送的方式，在訊息發送的程序中，涉及到了兩個執行緒——main 執行緒和 Sender 執行緒，以及一個執行緒共享變數——RecordAccumulator，

main 執行緒將訊息發送給 RecordAccumulator，Sender 執行緒不斷從 RecordAccumulator 中拉取訊息發送到 Kafka broker，

相關引數：
batch.size：只有資料積累到 batch.size 之后，sender 才會發送資料，
linger.ms：如果資料遲遲未達到 batch.size，sender 等待 linger.time 之后就會發送資料，

生產者設定攔截器

攔截器一：

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
/**
 * @ClassName TimeInterceptor
 * @Description 對于 producer 而言，interceptor 使得用戶在訊息發送前以及 producer 回呼邏輯前有機會
 *              對訊息做一些定制化需求，比如修改訊息等，producer 允許用戶指定多個 interceptor
 *              按序作用于同一條訊息從而形成一個攔截鏈.
 *              interceptor 可能被運行在多個執行緒中，因此在具體實作時用戶需要自行確保執行緒安全，
 *              另外倘若指定了多個 interceptor，則 producer 將按照指定順序呼叫它們，并僅僅
 *              是捕獲每個 interceptor 可能拋出的例外記錄到錯誤日志中而非在向上傳遞，這在使用程序中
 *              要特別留意，
 * @Author liuzhihui
 * @Date 15/1/2022 下午 9:06
 * @Version 1.0
 **/
public class TimeInterceptor implements ProducerInterceptor {
    /**
     * 獲取配置資訊和初始化資料時呼叫，
     */
    @Override
    public void configure(Map<String, ?> map) {
    }
    /**
     * 該方法封裝進 KafkaProducer.send 方法中，即它運行在用戶主執行緒中，
     * Producer 確保在訊息被序列化以及計算磁區前呼叫該方法，
     * 用戶可以在該方法中對訊息做任何操作，但最好保證不要修改訊息所屬的 topic
     * 和磁區，否則會影響目標磁區的計算，
     * @param producerRecord
     * @return
     */
    @Override
    public ProducerRecord onSend(ProducerRecord producerRecord) {
        // 需求：給資料加一個時間戳
        long time = System.currentTimeMillis();
        String value = (String) producerRecord.value();
        return new ProducerRecord(producerRecord.topic(),producerRecord.partition()
                ,producerRecord.key(), time+","+value);
    }
    /**
     * 該方法會在訊息從 RecordAccumulator 成功發送到 Kafka Broker 之后，或者在發送程序
     * 中失敗時呼叫，并且通常都是在 producer 回呼邏輯觸發之前，onAcknowledgement 運行在
     * producer 的 IO 執行緒中，因此不要在該方法中放入很重的邏輯，否則會拖慢 producer 的訊息
     * 發送效率，
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
    }
    /**
     * producer關閉的時候，呼叫
     */
    @Override
    public void close() {
    }
}

接下來使攔截器生效，

如果想攔截器僅針對部分訊息生效，可以在onSend入參中，判斷是不是想要發的topic

常見問題

1.Kafka 中的 ISR(InSyncRepli)、OSR(OutSyncRepli)、AR(AllRepli)代表什么？

ISR : 速率和leader相差低于10秒的follower的集合
OSR : 速率和leader相差大于10秒的follower
AR : 所有磁區的follower

2.Kafka 中的 HW、LEO 等分別代表什么？

LEO：指的是每個副本最大的 offset；
HW：指的是消費者能見到的最大的 offset，ISR 佇列中最小的 LEO，

3.Kafka 中是怎么體現訊息順序性的？

每個磁區內,每條訊息都有offset,所以只能在同一磁區內有序,但不同的磁區無法做到訊息順序性

4.Kafka 中的磁區器、序列化器、攔截器是否了解？它們之間的處理順序是什么？

攔截器>序列化器>磁區器

5.Kafka 生產者客戶端的整體結構是什么樣子的？使用了幾個執行緒來處理？分別是什么？

使用兩個執行緒:
main執行緒和sender執行緒
main執行緒會依次經過攔截器,序列化器,磁區器將資料發送到RecourdAccumlator(執行緒共享變數)
再由sender執行緒從RecourdAccumlator中拉取資料發送到kafka broker
相關引數：
batch.size：只有資料積累到batch.size之后，sender才會發送資料，
linger.ms：如果資料遲遲未達到batch.size，sender等待linger.time之后就會發送資料，

6.“消費組中的消費者個數如果超過 topic 的磁區，那么就會有消費者消費不到資料”這句
話是否正確？

對的,超過磁區數的消費者就不會再接收資料

7.消費者提交消費位移時提交的是當前消費到的最新訊息的 offset 還是 offset+1？

offset + 1
生產者發送資料offset是從0開始的
消費者消費的資料offset是從offset+1開始的

8.有哪些情形會造成重復消費？

先處理訊息，再提交offset，因為可能處理了訊息，但是因為kafka掛了或者其他原因，offset提交失敗，導致訊息重新消費

9.那些情景會造成訊息漏消費？

先提交offset，再消費，如果提交了，然后消費失敗了，就丟訊息  

10.當你使用 kafka-topics.sh 創建（洗掉）了一個 topic 之后，Kafka 背后會執行什么邏輯？

 1）會在 zookeeper 中的/brokers/topics 節點下創建一個新的 topic 節點
 如：/brokers/topics/first
 2）觸發 Controller 的監聽程式
 3）kafka Controller 負責 topic 的創建作業，并更新 metadata cache

11.topic 的磁區數可不可以增加？如果可以怎么增加？如果不可以，那又是為什么？

可增; 通過獲取當前可用磁區，投入到新的生產環境中使用

12.topic 的磁區數可不可以減少？如果可以怎么減少？如果不可以，那又是為什么？

不可減;先有的磁區資料難以處理

13.Kafka 有內部的 topic 嗎？如果有是什么？有什么所用？

__consumer_offsets：作用是保存 Kafka 消費者的位移資訊
__transaction_state：用來存盤事務日志訊息

14.Kafka 磁區分配的概念？

Kafka 有兩種分配策略，一是 RoundRobin，一是 Range，

15.簡述 Kafka 的日志目錄結構？

每一個磁區對應一個檔案夾,命名為topic-0,topic-1,每個檔案夾內有.index和.log檔案

16.如果我指定了一個 offset，Kafka Controller 怎么查找到對應的訊息？

通過二分查找法，找到index檔案，掃描index檔案，找到資料在log檔案中的地址

17.聊一聊 Kafka Controller 的作用？

負責kafka集群的上下線作業,所有topic的副本磁區分配和選舉leader作業

18.Kafka 中有那些地方需要選舉？這些地方的選舉策略又有哪些？

在ISR中需要選擇,選擇策略為先到先得

19.失效副本是指什么？有那些應對措施？

失效副本為速率比leader相差大于10秒的follower
將失效的follower先提出ISR
等速率接近leader10秒內,再加進ISR

20.Kafka 的哪些設計讓它有如此高的性能？

1.kafka是分布式的訊息佇列
2.對log檔案進行了segment,并對segment建立了索引
3.(對于單節點)使用了順序讀寫,速度可以達到600M/s
4.參考了zero拷貝,在os系統就完成了讀寫操作