来自多个线程的指标

Question

所以这似乎是一个非常常见的用例，也许我想多了，但我在保持来自多个线程的集中度量时遇到了问题。假设我有多个工作线程都在处理记录，并且每 1000 条记录我想吐出一些指标。现在我可以让每个线程记录单独的指标，然后获取吞吐量数字，但我必须手动将它们相加（当然时间界限并不准确）。下面是一个简单的例子：

public class Worker implements Runnable {

   private static int count = 0;
   private static long processingTime = 0;

   public void run() {
       while (true) {
          ...get record
          count++;
          long start = System.currentTimeMillis();
          ...do work
          long end = System.currentTimeMillis();
          processingTime += (end-start);
          if (count % 1000 == 0) {
              ... log some metrics
              processingTime = 0;
              count = 0;
          }
       }
    }
}

希望这有点道理。我也知道这两个静态变量可能是 AtomicInteger 和 AtomicLong 。 . .但也许不是。对人们有什么样的想法感兴趣。我曾考虑过使用原子变量并使用 ReeantrantReadWriteLock - 但我真的不希望指标停止处理流程（即指标对处理的影响应该非常小）。谢谢。

Answer 1

将实际处理任务转移到另一个线程可能是个好主意。这个想法是封装您的数据并快速将其传递给处理线程，从而最大限度地减少对正在执行有意义工作的线程的影响。

有一个小的切换争用，但这个成本通常比任何其他类型的同步要小很多，在许多情况下它应该是一个很好的候选者。我认为 M. Jessup 的解决方案与我的非常接近，但希望以下代码清楚地说明了这一点。

public class Worker implements Runnable {

   private static final Metrics metrics = new Metrics();

   public void run() {
      while (true) {
        ...get record
        long start = System.currentTimeMillis();
        ...do work
        long end = System.currentTimeMillis();
        // process the metric asynchronously
        metrics.addMetric(end - start);
     }
  }

  private static final class Metrics {
     // a single "background" thread that actually handles
     // processing
     private final ExecutorService metricThread = 
           Executors.newSingleThreadExecutor();
     // data (no synchronization needed)
     private int count = 0;
     private long processingTime = 0;

     public void addMetric(final long time) {
        metricThread.execute(new Runnable() {
           public void run() {
              count++;
              processingTime += time;
              if (count % 1000 == 0) {
                 ... log some metrics
                 processingTime = 0;
                 count = 0;
              }
           }
        });
      }
   }
}

Answer 2

我建议如果您不希望日志记录干扰处理，您应该有一个单独的日志工作线程，并让您的处理线程简单地提供某种类型的值对象，可以移交。在示例中，我选择了 LinkedBlockingQueue，因为它能够使用 offer() 阻塞很短的时间，并且您可以将阻塞推迟到从队列中提取值的另一个线程。您可能需要在 MetricProcessor 中增加逻辑以根据您的要求对数据等进行排序，但即使这是一个长时间运行的操作，它也不会阻止 VM 线程调度程序同时重新启动实际处理线程。

public class Worker implements Runnable {

  public void run() {
    while (true) {
      ... do some stuff
      if (count % 1000 == 0) {
        ... log some metrics
        if(MetricProcessor.getInstance().addMetrics(
            new Metrics(processingTime, count, ...)) {
          processingTime = 0;
          count = 0;
        } else {
          //the call would have blocked for a more significant
          //amount of time, here the results
          //could be abandoned or just held and attempted again
          //as a larger data set later
        }
      }
    }
  }
}

public class WorkerMetrics {
  ...some interesting data
  public WorkerMetrics(... data){
    ...
  }
  ...getter setters etc
}

public class MetricProcessor implements Runnable {
  LinkedBlockingQueue metrics = new LinkedBlockingQueue();
  public boolean addMetrics(WorkerMetrics m) {
    return metrics.offer(m); //This may block, but not for a significant amount of time.
  }

  public void run() {
    while(true) {
      WorkMetrics m = metrics.take(); //wait here for something to come in
      //the above call does all the significant blocking without
      //interrupting the real processing
      ...do some actual logging, aggregation, etc of the metrics
    }
  }
}

Answer 3

如果您依赖 count 的状态和 processingTime 的状态来同步，那么您将不得不使用 Lock。例如，当++count % 1000 == 0 为真时，您想评估当时处理时间的指标。

在这种情况下，使用 ReentrantLock 是有意义的。我不会使用 RRWL，因为实际上并没有发生纯读取的实例。它始终是一个读/写集。但是你需要锁定所有的

  count++
  processingTime += (end-start);
  if (count % 1000 == 0) {
      ... log some metrics
      processingTime = 0;
      count = 0;
  }

无论 count++ 是否会在该位置，您也需要锁定该位置。 最后，如果您使用的是 Lock，则不需要 AtomicLong 和 AtomicInteger。它只是增加了开销并且不是线程安全的。