https://github.com/apache/kafka/pull/16686

背景故事

這將會是一個很長的故事，很久很久以前，Kafka Consumer在關閉的時候有兩個特別的事情需要處理，第一個是ConsumerRebalanceListener必須被呼叫無論是否超時或是中斷，第二個則是要盡可能的把離開的請求送出去。聽起來是不是很莫來由？請聽小編一個一個說明：

ConsumerRebalanceListener是一個可以由使用者定義，並且會在partition的訂閱發生變化時被觸發，例如當你的consumer被分配了新的partition又或是它失去了手上的partition。由於在關閉的時候一定會釋放掉手上的partition，因此ConsumerRebalanceListener理論上是一定要被觸發，至少在classic consumer的手上這個邏輯是肯定的。但很不幸的，在新版async consumer的世界裡面，這件事情變得有一些些複雜，原來ConsumerRebalanceListener被執行其實有幾個要求，首先它必須是被關閉consumer的主執行緒所執行，為了讓這個條件成立，async consumer的主執行緒必須等待背景執行緒丟回來的請求，也就是會出現執行緒之間的等待行為，所以就延伸出主執行緒究竟要等到什麼時候？它該相信背景執行緒一定會丟請求回來嘛？而目前的做法就是盡量試試看，也就是很有可能在關閉async consumer的時候，ConsumerRebalanceListener會被忽略掉如果你運氣不好的話 ...

至於要把離開的請求送出去，這個在之前的文章有稍微提到，這裏簡單再說明一下，離開的請求是用來跟伺服器端說請把我這個consumer手上的任務分發給其他人，因此發送離開的請求是一個很重要的事情，它能讓我們保持讀取資料的速度。但是關聯到伺服器端的互動就會出現有可能很高的延遲，畢竟我們都不知道今天網路穩不穩、伺服器開不開心。因此大家可想而知這個會是一個需要取捨的設計，如果你堅持一定要送完才能關閉，那你的consumer就有可能卡死在關閉的狀態。但如果你選擇讓關閉這件事情可以超時或是中斷，那也就代表你可能會無法成功傳送離開的請求，進而導致某些資料一直沒有被重新分發和處理。classic consumer目前的處理邏輯是會努力盡量傳送看看，就算timeout=0它也會至少送一次看看，同時也會尊重中斷的訊號，如果送到一半被中斷，那classic consumer就會果斷宣告放棄，畢竟客戶最大他想中斷就讓他中斷吧。然而很可惜的事情是async consumer目前並沒有尊重中斷這件事情，也就是async consumer的主執行緒被中斷時，該中斷訊號並不會傳遞到背景執行緒，這會導致背景執行緒無法被中斷即時使用者真得非常想要這件事情發生 ...

解決辦法

同樣的既然知道原因，我們就能準確的對症下藥，首先主執行緒必須明確等到ConsumerRebalanceListener的請求出現並且執行完成，在這之後也必須等待背景執行緒將相關的狀態更新完成，為了避免被中斷所影響，在等到背景執行緒之前，我們必須先抑制中斷訊號以避免無法正確等待背景執行緒。針對第二點則是主執行緒在等待背景執行緒做清理的時候，如果收到了中斷訊號就必須確實的傳遞給背景執行緒，然後讓背景執行緒決定中斷後還有哪些事情需要完成，而主執行緒傳遞完中斷訊號後則是需要繼續等待，確保背景執行緒的清理完成和結束！

廣告

歡迎訂閱全臺最鬆散的開源社群源來適你，上面不定期會有各種開源的廢文。也歡迎參加全臺最鬆散的開源討論頻道，上面有一群網友一起在刷開源技術