Springboot整合kafka高階應用實戰分享

2022-08-14 18:03:30

深入應用

1.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）啟動資訊

1.2 訊息傳送

1.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的訊息。

1.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}

1.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注入

1.3 訊息消費

1.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這個程序被閒置

1.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()。

到此這篇關於Springboot整合kafka高階應用實戰分享的文章就介紹到這了,更多相關Springboot整合kafka 內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！