線上某服務 A 呼叫服務 B 介面完成一次交易，一次晚上的生產變更之後，系統監控發現服務 B 介面頻繁超時，後續甚至返回執行緒池耗盡錯誤 Thread pool is EXHAUSTED。因為服務 B 依賴外部介面，剛開始誤以為外部介面延時導致，所以臨時增加服務 B dubbo 執行緒池執行緒數量。配置變更之後，重啟服務，服務恢復正常。一段時間之後，服務 B 再次返回執行緒池耗盡錯誤。這次深入排查問題之後，才發現 Kafka 非同步傳送訊息阻塞了 dubbo 執行緒，從而導致呼叫超時。

一、問題分析

Dubbo 2.6.5，Kafak maven 0.8.0-beta1

服務 A 呼叫服務 B，收到如下錯誤：

2019-08-30 09:14:52,311 WARN method [%f [DUBBO] Thread pool is EXHAUSTED! Thread Name: DubboServerHandler-xxxx, Pool Size: 1000 (active: 1000, core: 1000, max: 1000, largest: 1000), Task: 6491 (completed: 5491), Executor status:(isShutdown:false, isTerminated:false, isTerminating:false), in dubbo://xxxx!, dubbo version: 2.6.0, current host: 127.0.0.1
 

可以看到當前 dubbo 執行緒池已經滿載執行，不能再接受新的呼叫。正常情況下 dubbo 執行緒可以很快完成任務，然後歸還到執行緒池中。由於執行緒執行的任務發生阻塞，消費者端呼叫超時。而服務提供者端由於已有執行緒被阻塞，執行緒池必須不斷建立新執行緒處理任務，直到執行緒數量達到最大數量，系統返回 Thread pool is EXHAUSTED。

執行緒任務長時間被阻塞可能原因有：

• 頻繁的 fullgc，導致系統暫停。
• 呼叫某些阻塞 API，如 socket 連線未設定超時時間導致阻塞。
• 系統內部死鎖

通過分析系統堆疊 dump 情況，果然發現所有 dubbo 執行緒都處於 WATTING 狀態。

下圖為應用堆疊 dump 日誌：

從堆疊日誌可以看到 dubbo 執行緒最後阻塞在 LinkedBlockingQueue#put ，而該阻塞發生在 Kafka 傳送訊息方法內。

這裡服務 B 需要使用 Kafka 傳送監控訊息，為了訊息傳送不影響主業務，這裡使用 Kafka 非同步傳送訊息。由於 Kafka 服務端最近更換了對外的埠，而服務 B Kafka 配置未及時變更。最後服務 B 修改配置，服務重新啟動，該問題得以解決。

二、Kafka 非同步模式

下面分析 Kafka 非同步傳送訊息阻塞的實際原因。

0.8.0 Kafka 預設使用同步模式傳送訊息，非同步傳送訊息需要設定producer.type=async屬性。同步模式需要等待 Kafka 將訊息傳送到訊息佇列，這個過程當然會阻塞主執行緒。而非同步模式最大的優點在於無需要等待 Kafka 這個傳送過程。

原本認為這裡的非同步是使用子執行緒去執行任務，但是 Kafka 非同步模式並非這樣。檢視 Kafka 官方文件producer,可以看到對非同步模式描述。

Batching is one of the big drivers of efficiency, and to enable batching the Kafka producer has an asynchronous mode that accumulates data in memory and sends out larger batches in a single request. The batching can be configured to accumulate no more than a fixed number of messages and to wait no longer than some fixed latency bound (say 100 messages or 5 seconds). This allows the accumulation of more bytes to send, and few larger I/O operations on the servers. Since this buffering happens in the client it obviously reduces the durability as any data buffered in memory and not yet sent will be lost in the event of a producer crash.

從上我們可以看到，Kafka 非同步模式將會把多條訊息打包一塊批量傳送到服務端。這種模式將會先把訊息放到記憶體佇列中，直到訊息到達一定數量（預設為 200）或者等待時間超限（預設為 5000ms）。

這麼做最大好處在於提高訊息傳送的吞吐量，減少網路 I/O。當然這麼做也存在明顯劣勢，如果生產者宕機，在記憶體中還未傳送訊息可能就會丟失。

下面從 kafka 原始碼分析這個阻塞過程。

三、Kafka 原始碼解析

Kafka 訊息傳送端採用如下配置:

        Properties props = new Properties();

        props.put("metadata.broker.list", "localhost:9092");
    // 選擇非同步傳送
        props.put("producer.type", "async");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("queue.buffering.max.messages","1");
        props.put("batch.num.messages","1");
        Producer<Integer, String> producer= new Producer(new ProducerConfig(props));
        producer.send(new KeyedMessage("test", "hello world"));

這裡設定 producer.type=async,從而使 Kafka 非同步傳送訊息。

send 方法原始碼如下：

ps: 這個版本 Kafka 原始碼採用 Scala 編寫，不過原始碼還是比較簡單，比較容易閱讀。

  def send(messages: KeyedMessage[K,V]*) {
    if (hasShutdown.get)
      throw new ProducerClosedException
    recordStats(messages)
    sync match {
      case true => eventHandler.handle(messages)
    // 由於  producer.type=async 非同步傳送
      case false => asyncSend(messages)
    }
  }

由於我們上面設定 producer.type=async，這裡將會使用 asyncSend 非同步傳送模式。

asyncSend 原始碼如下：

  private def asyncSend(messages: Seq[KeyedMessage[K,V]]) {
    for (message <- messages) {
      val added = config.queueEnqueueTimeoutMs match {
        case 0  =>
          queue.offer(message)
        case _  =>
          try {
            config.queueEnqueueTimeoutMs < 0 match {
    
            case true =>
              queue.put(message)
              true
            case _ =>
              queue.offer(message, config.queueEnqueueTimeoutMs, TimeUnit.MILLISECONDS)
            }
          }
          catch {
            case e: InterruptedException =>
              false
          }
      }
      if(!added) {
        producerTopicStats.getProducerTopicStats(message.topic).droppedMessageRate.mark()
        producerTopicStats.getProducerAllTopicsStats.droppedMessageRate.mark()
        throw new QueueFullException("Event queue is full of unsent messages, could not send event: " + message.toString)
      }else {
        trace("Added to send queue an event: " + message.toString)
        trace("Remaining queue size: " + queue.remainingCapacity)
      }
    }
  }

asyncSend 將會把訊息加入到 LinkedBlockingQueue 阻塞佇列中。這裡根據 config.queueEnqueueTimeoutMs引數使用不同方法。

當 config.queueEnqueueTimeoutMs=0，將會呼叫 LinkedBlockingQueue#offer，如果該佇列未滿，將會把元素插入佇列隊尾。如果佇列未滿，直接返回 false。所以如果此時佇列已滿，訊息不再會加入佇列中，然後 asyncSend 將會丟擲 QueueFullException 異常。

當 config.queueEnqueueTimeoutMs < 0,將會呼叫 LinkedBlockingQueue#put 加入元素，如果該佇列已滿，該方法將會一直被阻塞直到佇列存在可用空間。

當 config.queueEnqueueTimeoutMs > 0,將會呼叫 LinkedBlockingQueue#offer，這裡與上面不同之處在於設定超時時間，如果佇列已滿將會阻塞知道超時。

config.queueEnqueueTimeoutMs引數通過 queue.enqueue.timeout.ms 配置生效，預設為 -1。預設情況下 LinkedBlockingQueue 最大數量為 10000，可以通過設定 queue.buffering.max.messages 改變佇列最大值。

訊息放到佇列中後，Kafka 將會使用一個非同步執行緒不斷從佇列中獲取訊息，批量傳送訊息。

非同步處理訊息程式碼如下：

  private def processEvents() {
    var lastSend = SystemTime.milliseconds
    var events = new ArrayBuffer[KeyedMessage[K,V]]
    var full: Boolean = false

    // drain the queue until you get a shutdown command
    Stream.continually(queue.poll(scala.math.max(0, (lastSend + queueTime) - SystemTime.milliseconds), TimeUnit.MILLISECONDS))
                      .takeWhile(item => if(item != null) item ne shutdownCommand else true).foreach {
      currentQueueItem =>
        val elapsed = (SystemTime.milliseconds - lastSend)
        // check if the queue time is reached. This happens when the poll method above returns after a timeout and
        // returns a null object
        val expired = currentQueueItem == null
        if(currentQueueItem != null) {
          trace("Dequeued item for topic %s, partition key: %s, data: %s"
              .format(currentQueueItem.topic, currentQueueItem.key, currentQueueItem.message))
          events += currentQueueItem
        }

        // check if the batch size is reached
        full = events.size >= batchSize

        if(full || expired) {
          if(expired)
            debug(elapsed + " ms elapsed. Queue time reached. Sending..")
          if(full)
            debug("Batch full. Sending..")
          // if either queue time has reached or batch size has reached, dispatch to event handler
          tryToHandle(events)
          lastSend = SystemTime.milliseconds
          events = new ArrayBuffer[KeyedMessage[K,V]]
        }
    }
    // send the last batch of events
    tryToHandle(events)
    if(queue.size > 0)
      throw new IllegalQueueStateException("Invalid queue state! After queue shutdown, %d remaining items in the queue"
        .format(queue.size))
  }

這裡非同步執行緒將會不斷從佇列中獲取任務，一旦條件滿足，就會批量傳送任務。該條件為：

1. 批量訊息數量達到 200，可以設定 batch.num.messages 引數改變配置。
2. 等待時間到達最大的超時時間，預設為 5000ms，可以設定 queue.buffering.max.ms 改變改配置。

四、問題解決辦法

上面問題雖然通過更換 Kafka 正確地址解決，但是為了預防下次該問題再發生，可以採用如下方案：

1. 改變 config.queueEnqueueTimeoutMs預設配置，像這種系統監控日誌允許丟失，所以可以設定 config.queueEnqueueTimeoutMs=0。
2. 升級 Kafka 版本，最新版本 Kafka 使用 Java 重寫傳送端邏輯，不再使用阻塞佇列儲存訊息。

本文首發於：studyidea.cn/kafka…

歡迎關注我的公眾號：程式通事，獲得日常乾貨推送。如果您對我的專題內容感興趣，也可以關注我的部落格：studyidea.cn

相關推薦