深入應(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)答級(jí)別:多少個(gè)分區(qū)副本備份完成時(shí)向生產(chǎn)者發(fā)送ack確認(rèn)(可選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消費(fèi)者
group-id: javagroup # 默認(rèn)的消費(fèi)組ID
enable-auto-commit: true # 是否自動(dòng)提交offset
auto-commit-interval: 100 # 提交offset延時(shí)(接收到消息后多久提交offset)
auto-offset-reset: latest #earliest，latest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

2）啟動(dòng)信息

1.2 消息發(fā)送

1.2.1 發(fā)送類型

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

詳細(xì)代碼參考：AsyncProducer.java

消費(fèi)者使用：KafkaConsumer.java

1）同步發(fā)送

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

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

2）阻斷

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

docker-compose pause kafka-1 kafka-2

在swagger發(fā)送消息

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

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

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

3）注冊(cè)監(jiān)聽(tīng)

代碼參考： KafkaListener.java

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

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

查看控制臺(tái)，等待一段時(shí)間后，異步發(fā)送失敗的消息會(huì)被回調(diào)給注冊(cè)過(guò)的listener

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

啟動(dòng)kafka

docker-compose unpause kafka-1 kafka-2

再次發(fā)送消息時(shí)，同步異步均可以正常收發(fā)，并且監(jiān)聽(tīng)進(jì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 中，即注冊(cè)的Listener的消息。

1.2.2 序列化

消費(fèi)者使用：KafkaConsumer.java

1）序列化詳解

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

2）自定義序列化

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

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

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

//默認(rèn)調(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)一切正常。但是消費(fèi)端消息內(nèi)容不對(duì)！

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

怎么辦？

3）解碼

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

代碼參考：MyDeserializer.java，實(shí)現(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投放到哪個(gè)分區(qū)，也是順序消費(fèi)保障的基石。

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

1）驗(yàn)證默認(rèn)分區(qū)規(guī)則

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

消費(fèi)者代碼使用：PartitionConsumer.java

通過(guò)swagger訪問(wèn)setKey：

看控制臺(tái)：

再訪問(wèn)setPartition來(lái)設(shè)置分區(qū)號(hào)0來(lái)發(fā)送

看控制臺(tái)：

2）自定義分區(qū)

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

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

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

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

備注：

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

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

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

1.3 消息消費(fèi)

1.3.1 消息組別

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

1）代碼參考：GroupConsumer.java，Listener拷貝3份，分別賦予兩組group，驗(yàn)證分組消費(fèi)：

2）啟動(dòng)

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

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

4）消費(fèi)端閑置

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

驗(yàn)證方式：

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

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

解析：

group2可以消費(fèi)到1、2兩條消息

group1下有兩個(gè)消費(fèi)者，但是只分配給了 -1 ， -2這個(gè)進(jìn)程被閑置

1.3.2 位移提交

1）自動(dòng)提交

前面的案例中，我們?cè)O(shè)置了以下兩個(gè)選項(xiàng)，則kafka會(huì)按延時(shí)設(shè)置自動(dòng)提交

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

2）手動(dòng)提交

有些時(shí)候，我們需要手動(dòng)控制偏移量的提交時(shí)機(jī)，比如確保消息嚴(yán)格消費(fèi)后再提交，以防止丟失或重復(fù)。

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

代碼參考：MyOffsetConfig.java

通過(guò)在消費(fèi)端的Consumer來(lái)提交偏移量，有如下幾種方式：

代碼參考：MyOffsetConsumer.java

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

指定偏移量提交：offset()

3）重復(fù)消費(fèi)問(wèn)題

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

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

驗(yàn)證過(guò)程：

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

但是！重啟試試：

無(wú)論重啟多少次，不提交偏移量的消費(fèi)組，會(huì)重復(fù)消費(fèi)一遍?。。?/p>

再通過(guò)命令行查詢偏移量試試：

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

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

這就造成一個(gè)陷阱：

如果異步提交，針對(duì)偶爾出現(xiàn)的提交失敗，不進(jìn)行重試不會(huì)有太大問(wèn)題，因?yàn)槿绻峤皇∈且驗(yàn)榕R時(shí)問(wèn)題導(dǎo)致的，那么后續(xù)的提交總會(huì)有成功的。只要成功一次，偏移量就會(huì)提交上去。
但是！如果這是發(fā)生在關(guān)閉消費(fèi)者時(shí)的最后一次提交，就要確保能夠提交成功，如果還沒(méi)提交完就停掉了進(jìn)程。就會(huì)造成重復(fù)消費(fèi)！
因此，在消費(fèi)者關(guān)閉前一般會(huì)組合使用commitAsync()和commitSync()。

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

最新評(píng)論