深入應用

4.1 springboot-kafka

1）組態檔

  kafka:
    bootstrap-servers: 52.82.98.209:10903,52.82.98.209:10904
    producer: # producer 生產者
      retries: 0 # 重試次數
      acks: 1 # 應答級別:多少個磁區副本備份完成時向生產者發送ack確認(可選0、1、all/-1)
      batch-size: 16384 # 一次最多發送資料量
      buffer-memory: 33554432 # 生產端緩沖區大小
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer: # consumer消費者
      group-id: javagroup # 默認的消費組ID
      enable-auto-commit: true # 是否自動提交offset
      auto-commit-interval: 100 # 提交offset延時(接收到訊息后多久提交offset)
      auto-offset-reset: latest  #earliest，latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

2）啟動資訊

4.2 訊息發送

4.2.1 發送型別

KafkaTemplate呼叫send時默認采用異步發送，如果需要同步獲取發送結果，呼叫get方法

詳細代碼參考：AsyncProducer.java

消費者使用：KafkaConsumer.java

1）同步發送

        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("test", JSON.toJSONString(message));
				//注意，可以設定等待時間，超出后，不再等候結果
        SendResult<String, Object> result = future.get(3,TimeUnit.SECONDS);
        logger.info("send result:{}",result.getProducerRecord().value());

通過swagger發送，控制臺可以正常列印send result

2）阻斷

在服務器上，將kafka暫停服務

docker-compose pause kafka-1 kafka-2

在swagger發送訊息

調同步發送：請求被阻斷，一直等待，超時后回傳錯誤

而調異步發送的（默認發送介面），請求立刻回傳，

那么，異步發送的訊息怎么確認發送情況呢？？？往下看！

3）注冊監聽

代碼參考： KafkaListener.java

可以給kafkaTemplate設定Listener來監聽訊息發送情況，實作內部的對應方法

 kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {});

查看控制臺，等待一段時間后，異步發送失敗的訊息會被回呼給注冊過的listener

com.itheima.demo.config.KafkaListener:error!message={"message":"1","sendTime":1609920296374}

啟動kafka

docker-compose unpause kafka-1 kafka-2

再次發送訊息時，同步異步均可以正常收發，并且監聽進入success回呼

com.itheima.demo.config.KafkaListener$1:ok,message={"message":"1","sendTime":1610089315395}
com.itheima.demo.controller.PartitionConsumer:patition=1,message:[{"message":"1","sendTime":1610089315395}]

可以看到，在內部類 KafkaListener$1 中，即注冊的Listener的訊息，

4.2.2 序列化

消費者使用：KafkaConsumer.java

1）序列化詳解

• 前面用到的是Kafka自帶的字串序列化器（org.apache.kafka.common.serialization.StringSerializer）
• 除此之外還有：ByteArray、ByteBuffer、Bytes、Double、Integer、Long 等
• 這些序列化器都實作了介面 （org.apache.kafka.common.serialization.Serializer）
• 基本上，可以滿足絕大多數場景

2）自定義序列化

自己實作，實作對應的介面即可，有以下方法：

public interface Serializer<T> extends Closeable {
    default void configure(Map<String, ?> configs, boolean isKey) {
    }

  	//理論上，只實作這個即可正常運行
    byte[] serialize(String var1, T var2);

  	//默認調上面的方法
    default byte[] serialize(String topic, Headers headers, T data) {
        return this.serialize(topic, data);
    }

    default void close() {
    }
}

案例，參考: MySerializer.java

在yaml中配置自己的編碼器

value-serializer: com.itheima.demo.config.MySerializer

重新發送，發現：訊息發送端編碼回呼一切正常，但是消費端訊息內容不對！

com.itheima.demo.controller.KafkaListener$1:ok,message={"message":"1","sendTime":1609923570477}
com.itheima.demo.controller.KafkaConsumer:message:"{\"message\":\"1\",\"sendTime\":1609923570477}"

怎么辦？

3）解碼

發送端有編碼并且我們自己定義了編碼，那么接收端自然要配備對應的解碼策略

代碼參考：MyDeserializer.java，實作方式與編碼器幾乎一樣！

在yaml中配置自己的解碼器

value-deserializer: com.itheima.demo.config.MyDeserializer

再次收發，訊息正常

com.itheima.demo.controller.AsyncProducer$1:ok,message={"message":"1","sendTime":1609924855896}
com.itheima.demo.controller.KafkaConsumer:message:{"message":"1","sendTime":1609924855896}

4.2.3 磁區策略

磁區策略決定了訊息根據key投放到哪個磁區，也是順序消費保障的基石，

• 給定了磁區號，直接將資料發送到指定的磁區里面去
• 沒有給定磁區號，給定資料的key值，通過key取上hashCode進行磁區
• 既沒有給定磁區號，也沒有給定key值，直接輪循進行磁區
• 自定義磁區，你想怎么做就怎么做

1）驗證默認磁區規則

發送者代碼參考：PartitionProducer.java

消費者代碼使用：PartitionConsumer.java

通過swagger訪問setKey：

看控制臺：

再訪問setPartition來設定磁區號0來發送

看控制臺：

2）自定義磁區

你想自己定義規則，根據我的要求，把訊息投放到對應的磁區去？ 可以！

參考代碼：MyPartitioner.java , MyPartitionTemplate.java ,

發送使用：MyPartitionProducer.java

使用swagger，發送0開頭和非0開頭兩種key試一試！

備注：

自己定義config引數，比較麻煩，需要打破默認的KafkaTemplate設定

可以將KafkaConfiguration.java中的getTemplate加上@Bean注解來覆寫系統默認bean

這里為了避免混淆，采用@Autowire注入

4.3 訊息消費

4.3.1 訊息組別

發送者使用：KafkaProducer.java

1）代碼參考：GroupConsumer.java，Listener拷貝3份，分別賦予兩組group，驗證分組消費：

2）啟動

3）通過swagger發送2條訊息

• 同一group下的兩個消費者，在group1均分訊息
• group2下只有一個消費者，得到全部訊息

4）消費端閑置

注意磁區數與消費者數的搭配，如果 （ 消費者數 > 磁區數量 ），將會出現消費者閑置，浪費資源！

驗證方式：

停掉專案，刪掉test主題，重新建一個 ，這次只給它分配一個磁區，

重新發送兩條訊息，試一試

決議：

group2可以消費到1、2兩條訊息

group1下有兩個消費者，但是只分配給了 -1 ， -2這個行程被閑置

4.3.2 位移提交

1）自動提交

前面的案例中，我們設定了以下兩個選項，則kafka會按延時設定自動提交

enable-auto-commit: true # 是否自動提交offset
auto-commit-interval: 100  # 提交offset延時(接收到訊息后多久提交offset)

2）手動提交

有些時候，我們需要手動控制偏移量的提交時機，比如確保訊息嚴格消費后再提交，以防止丟失或重復，

下面我們自己定義配置，覆寫上面的引數

代碼參考：MyOffsetConfig.java

通過在消費端的Consumer來提交偏移量，有如下幾種方式：

代碼參考：MyOffsetConsumer.java

同步提交、異步提交：manualCommit() ，同步異步的差別，下面會詳細講到，

指定偏移量提交：offset()

3）重復消費問題

如果手動提交模式被打開，一定不要忘記提交偏移量，否則會造成重復消費！

代碼參考和對比：manualCommit() , noCommit()

驗證程序：

用km將test主題洗掉，新建一個test空主題，方便觀察訊息偏移
注釋掉其他Consumer的Component注解，只保留當前MyOffsetConsumer.java
啟動專案，使用swagger的KafkaProducer發送連續幾條訊息
留心控制臺，都能消費，沒問題：

但是！重啟試試：

無論重啟多少次，不提交偏移量的消費組，會重復消費一遍！！！

再通過命令列查詢偏移量試試：

4）經驗與總結

commitSync()方法，即同步提交，會提交最后一個偏移量，在成功提交或碰到無怯恢復的錯誤之前，commitSync()會一直重試，但是commitAsync()不會，

這就造成一個陷阱：
如果異步提交，針對偶爾出現的提交失敗，不進行重試不會有太大問題，因為如果提交失敗是因為臨時問題導致的，那么后續的提交總會有成功的，只要成功一次，偏移量就會提交上去，

但是！如果這是發生在關閉消費者時的最后一次提交，就要確保能夠提交成功，如果還沒提交完就停掉了行程，就會造成重復消費！

因此，在消費者關閉前一般會組合使用commitAsync()和commitSync()，
詳細代碼參考：MyOffsetConsumer.manualOffset()

本文由傳智教育博學谷 - 狂野架構師教研團隊發布
如果本文對您有幫助，歡迎關注和點贊；如果您有任何建議也可留言評論或私信，您的支持是我堅持創作的動力
轉載請注明出處！

標籤：Java

上一篇：域名+埠號 訪問minio服務問題

下一篇：網易二面：Kafka為什么吞吐量大、速度快？？