深入應(yīng)用

1.1 springboot-kafka

1）配置文件

kafka:
bootstrap-servers: 52.82.98.209:10903,52.82.98.209:10904
producer: # producer 生產(chǎn)者
retries: 0 # 重試次數(shù)
acks: 1 # 應(yīng)答級別:多少個分區(qū)副本備份完成時向生產(chǎn)者發(fā)送ack確認(可選0、1、all/-1)
batch-size: 16384 # 一次最多發(fā)送數(shù)據(jù)量
buffer-memory: 33554432 # 生產(chǎn)端緩沖區(qū)大小
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 消息發(fā)送

1.2.1 發(fā)送類型

KafkaTemplate調(diào)用send時默認采用異步發(fā)送，如果需要同步獲取發(fā)送結(jié)果，調(diào)用get方法

詳細代碼參考：AsyncProducer.java

消費者使用：KafkaConsumer.java

1）同步發(fā)送

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

通過swagger發(fā)送，控制臺可以正常打印send result

2）阻斷

在服務(wù)器上，將kafka暫停服務(wù)

docker-compose pause kafka-1 kafka-2

在swagger發(fā)送消息

調(diào)同步發(fā)送：請求被阻斷，一直等待，超時后返回錯誤

而調(diào)異步發(fā)送的（默認發(fā)送接口），請求立刻返回。

那么，異步發(fā)送的消息怎么確認發(fā)送情況呢？？？往下看！

3）注冊監(jiān)聽

代碼參考： KafkaListener.java

可以給kafkaTemplate設(shè)置Listener來監(jiān)聽消息發(fā)送情況，實現(xiàn)內(nèi)部的對應(yīng)方法

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

查看控制臺，等待一段時間后，異步發(fā)送失敗的消息會被回調(diào)給注冊過的listener

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

啟動kafka

docker-compose unpause kafka-1 kafka-2

再次發(fā)送消息時，同步異步均可以正常收發(fā)，并且監(jiān)聽進入success回調(diào)

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

可以看到，在內(nèi)部類 KafkaListener$1 中，即注冊的Listener的消息。

1.2.2 序列化

消費者使用：KafkaConsumer.java

1）序列化詳解

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

2）自定義序列化

自己實現(xiàn)，實現(xiàn)對應(yīng)的接口即可，有以下方法：

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

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

//默認調(diào)上面的方法
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

重新發(fā)送，發(fā)現(xiàn)：消息發(fā)送端編碼回調(diào)一切正常。但是消費端消息內(nèi)容不對！

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

怎么辦？

3）解碼

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

代碼參考：MyDeserializer.java，實現(xiàn)方式與編碼器幾乎一樣！

在yaml中配置自己的解碼器

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

再次收發(fā)，消息正常

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 分區(qū)策略

分區(qū)策略決定了消息根據(jù)key投放到哪個分區(qū)，也是順序消費保障的基石。

• 給定了分區(qū)號，直接將數(shù)據(jù)發(fā)送到指定的分區(qū)里面去
• 沒有給定分區(qū)號，給定數(shù)據(jù)的key值，通過key取上hashCode進行分區(qū)
• 既沒有給定分區(qū)號，也沒有給定key值，直接輪循進行分區(qū)
• 自定義分區(qū)，你想怎么做就怎么做

1）驗證默認分區(qū)規(guī)則

發(fā)送者代碼參考：PartitionProducer.java

消費者代碼使用：PartitionConsumer.java

通過swagger訪問setKey：

看控制臺：

再訪問setPartition來設(shè)置分區(qū)號0來發(fā)送

看控制臺：

2）自定義分區(qū)

你想自己定義規(guī)則，根據(jù)我的要求，把消息投放到對應(yīng)的分區(qū)去？ 可以！

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

發(fā)送使用：MyPartitionProducer.java

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

備注：

自己定義config參數(shù)，比較麻煩，需要打破默認的KafkaTemplate設(shè)置

可以將KafkaConfiguration.java中的getTemplate加上@Bean注解來覆蓋系統(tǒng)默認bean

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

1.3 消息消費

1.3.1 消息組別

發(fā)送者使用：KafkaProducer.java

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

2）啟動

3）通過swagger發(fā)送2條消息

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

4）消費端閑置

注意分區(qū)數(shù)與消費者數(shù)的搭配，如果 （ 消費者數(shù) > 分區(qū)數(shù)量 ），將會出現(xiàn)消費者閑置，浪費資源！

驗證方式：

停掉項目，刪掉test主題，重新建一個 ，這次只給它分配一個分區(qū)。

重新發(fā)送兩條消息，試一試

解析：

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

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

1.3.2 位移提交

1）自動提交

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

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

2）手動提交

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

下面我們自己定義配置，覆蓋上面的參數(shù)

代碼參考：MyOffsetConfig.java

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

代碼參考：MyOffsetConsumer.java

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

指定偏移量提交：offset()

3）重復(fù)消費問題

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

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

驗證過程：

用km將test主題刪除，新建一個test空主題。方便觀察消息偏移 注釋掉其他Consumer的Component注解，只保留當前MyOffsetConsumer.java 啟動項目，使用swagger的KafkaProducer發(fā)送連續(xù)幾條消息 留心控制臺，都能消費，沒問題：

但是！重啟試試：

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

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

4）經(jīng)驗與總結(jié)

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

這就造成一個陷阱：

如果異步提交，針對偶爾出現(xiàn)的提交失敗，不進行重試不會有太大問題，因為如果提交失敗是因為臨時問題導(dǎo)致的，那么后續(xù)的提交總會有成功的。只要成功一次，偏移量就會提交上去。
但是！如果這是發(fā)生在關(guān)閉消費者時的最后一次提交，就要確保能夠提交成功，如果還沒提交完就停掉了進程。就會造成重復(fù)消費！
因此，在消費者關(guān)閉前一般會組合使用commitAsync()和commitSync()。

到此這篇關(guān)于Springboot集成kafka高級應(yīng)用實戰(zhàn)分享的文章就介紹到這了,更多相關(guān)Springboot集成kafka 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論