背景

這是我們團隊負責的一個不太核心的服務。之前與外部交互時應外部要求由普通kafka集群改成加密kafka集群。我們是數據生產端。

改的過程中并跑上線，60%的請求耗時增加了2倍，也還是在百毫秒的量級可以接受。但是每次重啟的第一個請求要5s以上，會超過；運行過程中，一兩個月也會有一次超時。因為我們有三次重試，整體沒有影響成功率。

上線的時候我們問過網絡組，還專門請教過公司專業(yè)負責kafka的團隊。結論是：第一，這個慢是外部交互方的問題，不是咱們這邊可以處理的；第二，參數上也沒有什么可以調優(yōu)的。

我們團隊內部還是不信邪，調了幾個參數，加測之后上線了。頻繁度降到了現在的一兩個月一次超時，但是沒有根治。因為本身這個服務不是特別核心，本身外部是允許有一定失敗率的，而且現在實際上也沒有失敗，幾年內業(yè)務量也是很平穩(wěn)的：1分鐘4筆。

而我上班時間的狀態(tài)基本上是我站在兩個人中間，我目的是想問一個人問題，結果卻先要回答另外一個人的問題，這時候還會出現第四個人說別的事。這個優(yōu)先級排不上。但是心疼開發(fā)小哥哥，每一兩個月就要處理一下因為這件事引起的告警。雖然實際不影響，告警出來了，我們就要排查核對是否還是這個問題，并且確實通過重試將消息推送出去了。

所以本次利用周末，希望可以根治這個疑難雜癥，減少運維成本。

思路

前期已經明確了這個外部的加密集群建立連接和數據傳輸速度都慢于之前的普通集群。之所以第一次慢和每一兩個月會慢一次都是連接斷開重連造成的。之前我們進行過參數調優(yōu)，調優(yōu)做的就是因為1分鐘4筆請求，線上以最小部署單元3臺機器部署，每臺機器1分鐘預計處理一筆請求。根據這個數據調整了空閑自動斷開連接的時間間隔，保證連接不會因為空閑自動斷開。線上驗證有效，也側面證實了是連接過程慢引起的超時。

因為建立連接過程慢，這個主要是外部提供的集群就是如此。既然目前并不影響實際發(fā)送成功率。人家代表的是大佬，我們也不好太強硬的去推他們解決。所以我的思路有兩個：

第一，探索將建立連接與發(fā)送數據分離的可行性：程序啟動后先將連接建立好再提供服務。如果生產端是這樣實現的。那也許還可以進行連接自動探測，如果連接斷開則自動重連，不要等發(fā)送數據時再發(fā)現連接已斷開。

第二，其實第一種思路的可行性渺茫，只是需要驗證一下自己的想法。一般的這種消息中間件，消費端是這樣實現的。但是生產端采用了更簡單的方式：讀寫數據的時候再探測連接是否可用，不可用則重新建立連接。這種用在發(fā)送本來就是異步的，對發(fā)送延遲本身敏感度也不高的場景。生產端本來就是這種場景，并且通過測試實際上也確實是在發(fā)送時建立的第一次連接。kafka生產端原本就是這種設計的可能性極大。如果是這種情況，那就在生產端真正使用異步，給調用方返回“受理成功”，保證調用方不超時。自己再通過接受回調保證實際的成功。

這個事情真要做，還有兩個隱形需求：

1、因為外部有需求，數據可以偶爾少發(fā)，但是不能重復發(fā)送。所以不能使用業(yè)務級別的數據發(fā)送來實現探測功能。重試也要保證上條確實沒有收到。

2、改造不能太大，研發(fā)成本要小。

過程

因為我在網上搜到的這方面都是入門級，沒有什么解決這個問題的相關資料。所以采用的主要方法是讀源碼和官方文檔。當然，本文的方法是有前提知識儲備基礎的。就是《白話TCP/IP原理》系列的相關內容:https://mp.weixin.qq.com/s/Y2k3AW2ZjWbB1w63gsSRag

步驟一，查詢版本特性

我們目前用到的kafka客戶端版本是

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.5.8.RELEASE</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.5.0</version>
</dependency>

spring-kafka對應的官網的大版本是2.5，所以先點開了2.5.17.RELEASE對應的參考文檔。看到一句有用信息：

The default consumer and producer factories can now invoke a callback whenever a consumer or producer is created or closed.

默認消費者和生產者工廠現在已經可以在生產者和消費者創(chuàng)建和關閉時引發(fā)一個回調。耗時的連接建立過程是可以監(jiān)聽的，我們可以通過打日志進行監(jiān)控。

步驟二，查源碼

首先我們看一下類圖，看不清楚沒有關系?？催@里就好：

首先發(fā)現Producer、Consumer和Sender都是通過KafkaClient（接口），也就是NetworkClient（實現類）進行網絡活動的。其次發(fā)現NetworkClient是在傳輸層和應用層之間起了一個緩沖的作用，解耦了各個部件。

Producer、Consumer和AdminClient主要管理requests；NetworkClient主要管理connection；Selector主要管理sockets channel。這些被管理對象我在之前的網絡系列里都講過。

如果不看代碼，我站在設計者角度結合類圖猜想：生產端實際使用的是KafkaTemplate的send方法，具體的參數都是由DefaultKafkaProducerFactory接收。實際上連接的建立是Producer類進行。而在Producer類依賴于NetworkClient。而實際上進行連接應該在Sender類。Sender是一個Runnable異步線程來做，那實際建立連接的是run方法中。

我跟蹤源碼驗證了猜想。NetworkClient里有個initiateConnect的私有方法，是建立連接用的，跟蹤它就可以知道調用的地方。跟蹤下來，主要入口在NetworkClient的poll方法，注釋如下：

 /**
     * Do actual reads and writes to sockets.
     *
     * @param timeout The maximum amount of time to wait (in ms) for responses if there are none immediately,
     *                must be non-negative. The actual timeout will be the minimum of timeout, request timeout and
     *                metadata timeout
     * @param now The current time in milliseconds
     * @return The list of responses received
     */
    @Override
    public List<ClientResponse> poll(long timeout, long now) {

人家明確說了是讀寫時才會調用。證實了思路一不可行。

步驟三，查自身的代碼

按照思路二，進行異步化。本身生產端就應該是異步的，為什么異步沒有生效呢？結合KafkaTemplate的send方法源代碼和項目中自己寫的代碼。異步部分大體是這樣：

 SettableListenableFuture future = new SettableListenableFuture();
        future.set("OK");
        future.get();
        future.addCallback((sendResult) -> {
            try {
                System.out.println("成功");
            } catch (Exception e) {
            }
        }, r -> {
            System.out.println("失敗");
        });
        System.out.println("============end==============");

就是說KafkaTemplate的異步是靠使用SettableListenableFuture實現的，實際上它的set方法會馬上觸發(fā)callback，是同步的。代碼是先同步調用set，并且還手動調用了get（這個方法會等待直到返回結果）。所以整體是同步的?；蛘咧苯舆@么看，future實現異步要有一個Callable或者Runnable的線程方法，人家SettableListenableFuture第一行源碼就禁用了Callable。這個我看了2.5.17.RELEASE這個更高版本的spring-kafka，實現沒有做更改。

也就是說spring-kafka自身起碼在2.5.X版本里異步沒有起到作用。

問題清楚了修改也很簡單，比如可以加個異步注解將整個發(fā)送方法做異步，重試等邏輯也放到這個方法中。給調用方只返回受理成功。具體怎么解決交給開發(fā)小哥哥。

總結

幸虧我上周已經提前規(guī)劃好周一要休假。否則現在都2點半了明天上班也沒精神。主要時間花在異步不生效的問題上。其實排查異步不生效的思路是很簡單清晰的。耗時長是因為：第一，不敢相信spring官方實現的，竟然使用異步的代碼實際效果沒有異步；第二，關于異步我在網上搜索了一下，都是按照項目中配置的那樣。官方這樣說，大家這樣說，我總得考慮是不是自己搞錯了。

所以我反復的驗證、反復的debug之后也不敢下結論。仔細研究了源碼仍然不敢下結論。直到終于搜索到一篇文章說要實現異步除了要使用addCallback之外還要加異步標簽。人間清醒的我，馬上意識到文章實際用了兩種不同方法實現異步。作者之所以認為這是一個方法的兩個部分大概也是發(fā)現其實spring-kafka的異步沒好使吧。

到此這篇關于Spring Kafka中如何通過參數配置解決超時問題的文章就介紹到這了,更多相關Spring Kafka參數配置解決超時內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論