Контроль размера очередей цикла событий Netty

Question

Мы реализовали мониторинг очередей циклов событий Netty, чтобы понять проблемы с некоторыми из наших модулей Netty. Монитор использует метод io.netty.util.concurrent.SingleThreadEventExecutor#pendingTasks, который работает для большинства модулей, но для модуля, который обрабатывает несколько тысяч HTTP-запросов в секунду, кажется, что он висел или очень медленный. Теперь я понимаю, что документы строго указывают, что это может быть проблемой, и я чувствую себя довольно хромой ... поэтому я ищу другой способ реализовать этот монитор.

Вы можете увидеть старый код здесь: https://github.com/outbrain/ob1k/blob/6364187b30cab5b79d64835131d9168c754f3c09/ob1k-core/src/main/java/com/outbrain/ob1k/common/metrics/NettyQueuesGaugeBuilder.java

public static void registerQueueGauges(final MetricFactory factory, final EventLoopGroup elg, final String componentName) { 

    int index = 0; 
    for (final EventExecutor eventExecutor : elg) { 
     if (eventExecutor instanceof SingleThreadEventExecutor) { 
     final SingleThreadEventExecutor singleExecutor = (SingleThreadEventExecutor) eventExecutor; 
     factory.registerGauge("EventLoopGroup-" + componentName, "EventLoop-" + index, new Gauge<Integer>() { 
      @Override 
      public Integer getValue() { 
      return singleExecutor.pendingTasks(); 
      } 
     }); 

     index++; 
     } 
    } 
    } 

Мой вопрос, есть лучший способ контролировать размеры очередей?

Это может быть весьма полезной метрикой, так как ее можно использовать для понимания латентности, а также для использования в некоторых случаях для обеспечения противодавления.

Answer 1

Возможно, вам необходимо будет отслеживать изменения как задачи, добавленные и удаленные из экземпляров SingleThreadEventExecutor.

Для этого вы можете создать класс, который обертывает и/или расширяет SingleThreadEventExecutor. Тогда у вас будет java.util.concurrent.atomic.AtomicInteger, который вы бы назвали incrementAndGet() каждый раз при добавлении новой задачи и decrementAndGet() каждый раз, когда вы удаляете/заканчиваете.

Тогда AtomicInteger предоставит вам текущее количество ожидающих задач. Возможно, вы могли бы переопределить pendingTasks(), чтобы использовать это значение (хотя будьте осторожны - я не 100%, у которого не было бы побочных эффектов).

Это добавило бы немного накладных расходов для каждой выполняемой задачи, но потребовало бы, чтобы количество ожидающих задач находилось рядом с постоянной скоростью.

Недостатком этого является, конечно, что он более инвазивен, чем то, что вы сейчас делаете, так как вам нужно настроить приложение для использования разных исполнителей событий.

NB. это всего лишь предложение о том, как обойти эту проблему - я специально не делал этого с Netty. Хотя в прошлом я делал это с другим кодом.