Springboot集成kafka高級應用實戰分享

Posted on by

深入應用

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 內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：