來自多個線程的指標

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

如果您依賴count的狀態和processingTime的狀態同步，那麼您將不得不使用Lock。例如，如果++count % 1000 == 0爲真，那麼您希望評估此時的processingTime指標。

對於這種情況，使用ReentrantLock是有意義的。我不會使用RRWL，因爲實際上並不存在純粹讀取發生的實例。它始終是一個讀/寫設置。但你將需要鎖定所有的

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

無論計數++是否將在該位置，您將需要鎖定這一點。 最後，如果您使用鎖定，則不需要AtomicLong和AtomicInteger。它只是增加了開銷，並沒有更多的線程安全。

Answer 3

我會建議，如果你不想讓日誌干擾處理，你應該有一個單獨的日誌工作線程，並讓你的處理線程只是提供一些類型的值對象，可以切換。在這個例子中，我選擇了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 
    } 
    } 
}