一、原因分析

消息重復(fù)消費(fèi)的根本原因都在于：已經(jīng)消費(fèi)了數(shù)據(jù)，但是offset沒(méi)有成功提交。

其中很大一部分原因在于發(fā)生了再均衡。

1）消費(fèi)者宕機(jī)、重啟等。導(dǎo)致消息已經(jīng)消費(fèi)但是沒(méi)有提交offset。

2）消費(fèi)者使用自動(dòng)提交offset，但當(dāng)還沒(méi)有提交的時(shí)候，有新的消費(fèi)者加入或者移除，發(fā)生了rebalance（再平衡）。再次消費(fèi)的時(shí)候，消費(fèi)者會(huì)根據(jù)提交的偏移量來(lái)，于是重復(fù)消費(fèi)了數(shù)據(jù)。

3）消息處理耗時(shí)，或者消費(fèi)者拉取的消息量太多，處理耗時(shí)，超過(guò)了max.poll.interval.ms的配置時(shí)間，導(dǎo)致認(rèn)為當(dāng)前消費(fèi)者已經(jīng)死掉，觸發(fā)再均衡。

4)? 每次拉取的消息記錄數(shù)max.poll.records為100，poll最大拉取間隔max.poll.interval.ms為 300s，消息處理過(guò)于耗時(shí)導(dǎo)致時(shí)長(zhǎng)大于了這個(gè)值，導(dǎo)致再均衡發(fā)生重復(fù)消費(fèi)

二、解決方案

由于網(wǎng)絡(luò)問(wèn)題，重復(fù)消費(fèi)不可避免，因此，消費(fèi)者需要實(shí)現(xiàn)消費(fèi)冪等。方案：

1、消息表

2、數(shù)據(jù)庫(kù)唯一索引

3、緩存消費(fèi)過(guò)的消息id

4、手動(dòng)提交office

5、減少每次拉取的消息記錄數(shù)和增大poll之間的時(shí)間間隔

6、拉取到消息之后異步處理（保證成功消費(fèi)）

提高消費(fèi)者的處理速度。例如：對(duì)消息處理中比較耗時(shí)的步驟可通過(guò)異步的方式進(jìn)行處理、利用多線程處理等。在縮短單條消息消費(fèi)的同時(shí)，根據(jù)實(shí)際場(chǎng)景可將max.poll.interval.ms值設(shè)置大一點(diǎn)，避免不必要的Rebalance?？筛鶕?jù)實(shí)際消息速率適當(dāng)調(diào)小max.poll.records的值。　　　

三、新版kafka的broker冪等性具體實(shí)現(xiàn)原理

kafka新版本已經(jīng)在broker中保證了接收消息的冪等性（比如2.4版本），只需在生產(chǎn)者加上參數(shù) props.put(“enable.idempotence”, true) 即可，默認(rèn)是false不開(kāi)啟。

kafka每次發(fā)送消息會(huì)生成PID和Sequence Number，并將這兩個(gè)屬性一起發(fā)送給broker，broker會(huì)將PID和Sequence Number跟消息綁定一起存起來(lái)，下次如果生產(chǎn)者重發(fā)相同消息，broker會(huì)檢查PID和Sequence Number，如果相同不會(huì)再接收。

PID：每個(gè)新的 Producer 在初始化的時(shí)候會(huì)被分配一個(gè)唯一的 PID，這個(gè)PID對(duì)用戶完全是透明的。生產(chǎn)者如果重啟則會(huì)生成新的PID。

Sequence Number：對(duì)于每個(gè) PID，該 Producer 發(fā)送到每個(gè) Partition 的數(shù)據(jù)都有對(duì)應(yīng)的序列號(hào)，這些序列號(hào)是從0開(kāi)始單調(diào)遞增的。

四、ACK可靠性保證

為保證producer發(fā)送的數(shù)據(jù)，能可靠的發(fā)送到指定的topic，topic的每個(gè)partition收到producer發(fā)送的數(shù)據(jù)后，都需要向producer發(fā)送ack（acknowledgement確認(rèn)收到），如果producer收到ack，就會(huì)進(jìn)行下一輪的發(fā)送，否則重新發(fā)送數(shù)據(jù)。

1、ack應(yīng)答級(jí)別

對(duì)于某些不太重要的數(shù)據(jù)，對(duì)數(shù)據(jù)的可靠性要求不是很高，能夠容忍數(shù)據(jù)的少量丟失，沒(méi)必要等ISR中的follower全部接收成功。所以Kafka為用戶提供了三種可靠性級(jí)別，用戶根據(jù)對(duì)可靠性和延遲的要求進(jìn)行權(quán)衡，選擇以下的配置。

acks參數(shù)配置：

• 0：這一操作提供了一個(gè)最低的延遲，partition的leader接收到消息還沒(méi)有寫入磁盤就已經(jīng)返回ack，當(dāng)leader故障時(shí)有可能丟失數(shù)據(jù)；
• 1： partition的leader落盤成功后返回ack，如果在follower同步成功之前l(fā)eader故障，那么將會(huì)丟失數(shù)據(jù)；
• -1（all）： partition的leader和follower全部落盤成功后才返回ack。但是如果在follower同步完成后，broker發(fā)送ack之前，leader發(fā)生故障，那么會(huì)造成數(shù)據(jù)重復(fù)。

2、常見(jiàn)配置

• fetch.min.byte：配置Consumer一次拉取請(qǐng)求中能從Kafka中拉取的最小數(shù)據(jù)量，默認(rèn)為1B，如果小于這個(gè)參數(shù)配置的值，就需要進(jìn)行等待，直到數(shù)據(jù)量滿足這個(gè)參數(shù)的配置大小。調(diào)大可以提交吞吐量，但也會(huì)造成延遲
• fetch.max.bytes，一次拉取數(shù)據(jù)的最大數(shù)據(jù)量，默認(rèn)為52428800B，也就是50M，但是如果設(shè)置的值過(guò)小，甚至小于每條消息的值，實(shí)際上也是能消費(fèi)成功的
• fetch.wait.max.ms，若是不滿足fetch.min.bytes時(shí)，等待消費(fèi)端請(qǐng)求的最長(zhǎng)等待時(shí)間，默認(rèn)是500ms
• max.poll.records，單次poll調(diào)用返回的最大消息記錄數(shù)，如果處理邏輯很輕量，可以適當(dāng)提高該值。一次從kafka中poll出來(lái)的數(shù)據(jù)條數(shù),max.poll.records條數(shù)據(jù)需要在在session.timeout.ms這個(gè)時(shí)間內(nèi)處理完，默認(rèn)值為500
• consumer.poll(100)?，100 毫秒是一個(gè)超時(shí)時(shí)間，一旦拿到足夠多的數(shù)據(jù)（fetch.min.bytes 參數(shù)設(shè)置），consumer.poll(100)會(huì)立即返回 ConsumerRecords<String, String> records。如果沒(méi)有拿到足夠多的數(shù)據(jù)，會(huì)阻塞100ms，但不會(huì)超過(guò)100ms就會(huì)返回
• max.poll.interval.ms，兩次拉取消息的間隔，默認(rèn)5分鐘；通過(guò)消費(fèi)組管理消費(fèi)者時(shí)，該配置指定拉取消息線程最長(zhǎng)空閑時(shí)間，若超過(guò)這個(gè)時(shí)間間隔沒(méi)有發(fā)起poll操作，則消費(fèi)組認(rèn)為該消費(fèi)者已離開(kāi)了消費(fèi)組，將進(jìn)行再均衡操作（將分區(qū)分配給組內(nèi)其他消費(fèi)者成員）

若超過(guò)這個(gè)時(shí)間則報(bào)如下異常：

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already
 rebalanced and assigned the partitions to another member. This means that the time between subsequent calls 
to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is 
spending too much time message processing. You can address this either by increasing the session timeout or by
 reducing the maximum size of batches returned in poll() with max.poll.records.　

即：無(wú)法完成提交，因?yàn)榻M已經(jīng)重新平衡并將分區(qū)分配給另一個(gè)成員。這意味著對(duì)poll()的后續(xù)調(diào)用之間的時(shí)間比配置的max.poll.interval.ms長(zhǎng)，這通常意味著poll循環(huán)花費(fèi)了太多的時(shí)間來(lái)處理消息。

可以通過(guò)增加max.poll.interval.ms來(lái)解決這個(gè)問(wèn)題，也可以通過(guò)減少在poll()中使用max.poll.records返回的批的最大大小來(lái)解決這個(gè)問(wèn)題

max.partition.fetch.bytes：該屬性指定了服務(wù)器從每個(gè)分區(qū)返回給消費(fèi)者的最大字節(jié)數(shù)，默認(rèn)為 1MB。

session.timeout.ms：消費(fèi)者在被認(rèn)為死亡之前可以與服務(wù)器斷開(kāi)連接的時(shí)間，默認(rèn)是 3s，將觸發(fā)再均衡操作，對(duì)于每一個(gè)Consumer Group，Kafka集群為其從Broker集群中選擇一個(gè)Broker作為其Coordinator。Coordinator主要做兩件事：

1. 維持Group成員的組成。這包括加入新的成員，檢測(cè)成員的存活性，清除不再存活的成員。

2. 協(xié)調(diào)Group成員的行為。

3、poll機(jī)制

①：每次poll的消息處理完成之后再進(jìn)行下一次poll，是同步操作

②：每次poll之前檢查是否可以進(jìn)行位移提交，如果可以，那么就會(huì)提交上一次輪詢的位移

③：每次poll時(shí)，consumer都將嘗試使用上次消費(fèi)的offset作為起始o(jì)ffset，然后依次拉取消息

④：poll(long timeout)，timeout指等待輪詢緩沖區(qū)的數(shù)據(jù)所花費(fèi)的時(shí)間，單位是毫秒文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-772596.html

到了這里，關(guān)于JAVA面試題分享一百六十二：Kafka消息重復(fù)消費(fèi)問(wèn)題？的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！