重復消費的原因
消息重復消費的根本原因都在於：已經消費了數據，但是offset沒有成功提交。
其中很大一部分原因在於發生了再均衡。
1）消費者宕機、重啟等。導致消息已經消費但是沒有提交offset。
2）消費者使用自動提交offset，但當還沒有提交的時候，有新的消費者加入或者移除，發生了rebalance。再次消費的時候，消費者會根據提交的偏移量來，於是重復消費了數據。
3）消息處理耗時，或者消費者拉取的消息量太多，處理耗時，超過了max.poll.interval.ms的配置時間，導致認為當前消費者已經死掉，觸發再均衡。
重復消費的解決方案
由於網絡問題，重復消費不可避免，因此，消費者需要實現消費冪等。
方案：
①：消息表
②：數據庫唯一索引
③：緩存消費過的消息id
提高消費者的處理速度。例如：對消息處理中比較耗時的步驟可通過異步的方式進行處理、利用多線程處理等。在縮短單條消息消費的同時，根據實際場景可將max.poll.interval.ms值設置大一點，避免不必要的Rebalance。可根據實際消息速率適當調小max.poll.records的值。
重復消費問題：
每次拉取的消息記錄數max.poll.records為100，poll最大拉取間隔max.poll.interval.ms為 300s，消息處理過於耗時導致時長大於了這個值，導致再均衡發生重復消費
解決辦法：
①：減少每次拉取的消息記錄數和增大poll之間的時間間隔
②：拉取到消息之后異步處理（保證成功消費）
注意：kafka新版本已經在broker中保證了接收消息的冪等性（比如2.4版本），只需在生產者加上參數 props.put(“enable.idempotence”, true) 即可，默認是false不開啟。
新版kafka的broker冪等性具體實現原理：
kafka每次發送消息會生成PID和Sequence Number，並將這兩個屬性一起發送給broker，broker會將PID和Sequence Number跟消息綁定一起存起來，下次如果生產者重發相同消息，broker會檢查PID和 Sequence Number，如果相同不會再接收。
PID：每個新的 Producer 在初始化的時候會被分配一個唯一的 PID，這個PID對用戶完全是透明的。生產者如果重啟則會生成新的PID。
Sequence Number：對於每個 PID，該 Producer 發送到每個 Partition 的數據都有對應的序列號，這些序列號是從0開始單調遞增的。
常見配置
fetch.min.byte：配置Consumer一次拉取請求中能從Kafka中拉取的最小數據量，默認為1B，如果小於這個參數配置的值，就需要進行等待，直到數據量滿足這個參數的配置大小。調大可以提交吞吐量，但也會造成延遲
fetch.max.bytes，一次拉取數據的最大數據量，默認為52428800B，也就是50M，但是如果設置的值過小，甚至小於每條消息的值，實際上也是能消費成功的
fetch.wait.max.ms，若是不滿足fetch.min.bytes時，等待消費端請求的最長等待時間，默認是500ms
max.poll.records，單次poll調用返回的最大消息記錄數，如果處理邏輯很輕量，可以適當提高該值。一次從kafka中poll出來的數據條數,max.poll.records條數據需要在在session.timeout.ms這個時間內處理完，默認值為500
consumer.poll(100) ，100 毫秒是一個超時時間，一旦拿到足夠多的數據（fetch.min.bytes 參數設置），consumer.poll(100)會立即返回 ConsumerRecords<String, String> records。如果沒有拿到足夠多的數據，會阻塞100ms，但不會超過100ms就會返回
max.poll.interval.ms，兩次拉取消息的間隔，默認5分鍾；通過消費組管理消費者時，該配置指定拉取消息線程最長空閑時間，若超過這個時間間隔沒有發起poll操作，則消費組認為該消費者已離開了消費組，將進行再均衡操作（將分區分配給組內其他消費者成員）
若超過這個時間則報如下異常：
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.
即：無法完成提交，因為組已經重新平衡並將分區分配給另一個成員。這意味着對poll()的后續調用之間的時間比配置的max.poll.interval.ms長，這通常意味着poll循環花費了太多的時間來處理消息。
可以通過增加max.poll.interval.ms來解決這個問題，也可以通過減少在poll()中使用max.poll.records返回的批的最大大小來解決這個問題
max.partition.fetch.bytes：該屬性指定了服務器從每個分區返回給消費者的最大字節數，默認為 1MB。
session.timeout.ms：消費者在被認為死亡之前可以與服務器斷開連接的時間，默認是 3s，將觸發再均衡操作
對於每一個Consumer Group，Kafka集群為其從Broker集群中選擇一個Broker作為其Coordinator。Coordinator主要做兩件事：
-
維持Group成員的組成。這包括加入新的成員，檢測成員的存活性，清除不再存活的成員。
-
協調Group成員的行為。
poll機制
①：每次poll的消息處理完成之后再進行下一次poll，是同步操作
②：每次poll之前檢查是否可以進行位移提交，如果可以，那么就會提交上一次輪詢的位移
③：每次poll時，consumer都將嘗試使用上次消費的offset作為起始offset，然后依次拉取消息
④：poll(long timeout)，timeout指等待輪詢緩沖區的數據所花費的時間，單位是毫秒