單元測試必須手動中斷的異步計算

Question

我有一個類可以異步記錄eyetracking數據。有記錄過程的方法有start和stop。數據收集在一個集合中，只有在記錄線程完成其工作時才能訪問集合。它基本上封裝了所有的線程和同步，所以我的庫的用戶不必這樣做。

全副縮短代碼（仿製藥和錯誤處理省略）：

public class Recorder { 
    private Collection accumulatorCollection; 
    private Thread recordingThread; 

    private class RecordingRunnable implements Runnable { 
    ... 

    public void run() { 
     while(!Thread.currentThread().isInterrupted()) { 
     // fetch data and collect it in the accumulator 
     synchronized(acc) { acc.add(Eyetracker.getData()) } 
     } 
    } 
    } 

    public void start() { 
    accumulatorCollection = new Collection(); 
    recordingThread = new Thread(new RecordingRunnable(accumulatorCollection)); 
    recordingThread.start(); 
    } 

    public void stop() { 
    recordingThread.interrupt(); 
    } 

    public void getData() { 
    try { 
     recordingThread.join(2000); 
     if(recordingThread.isAlive()) { throw Exception(); } 
    } 
    catch(InterruptedException e) { ... } 

    synchronized(accumulatorCollection) { return accumulatorCollection; } 
    } 
} 

的用法很簡單：

recorder.start(); 
... 
recorder.stop(); 
Collection data = recorder.getData(); 

我對整個事情的問題是如何對其進行測試。目前，我做這樣的：

recorder.start(); 
Thread.sleep(50); 
recorder.stop(); 
Collection data = recorder.getData(); 
assert(stuff); 

這工作，但它具有不確定性，並減慢測試套件相當多的（我畫這些測試，集成測試，所以他們必須單獨運行繞過這個問題）。

有沒有更好的方法？

Answer 1

使用CountDownLatch有更好的方法。

測試的不確定性部分來自兩個變量在時間莖你不佔：

• 創建和啓動一個線程需要時間和線程可能沒有開始執行的可運行時Thread.start()回報（runnable將被執行，但可能稍後）。
• 停止/中斷將打斷Runnable中的while循環，但不會立即執行，可能稍後。

這是CountDownLatch進來的地方：它爲您提供有關另一個線程執行的位置的精確信息。例如。讓第一個線程等待鎖存器，而第二個線程作爲可運行的最後一條語句「倒計時」，現在第一個線程知道runnable已完成。 CountDownLatch也充當同步器：無論第二個線程寫入內存，現在都可以被第一個線程讀取。

而不是使用中斷，你也可以使用一個volatile布爾值。讀取volatile變量的任何線程都可以保證看到任何其他線程設置的最後一個值。

一個CountDownLatch也可以給出一個超時，其對可掛測試是非常有用的：如果你要等待很長時間，則可以中止整個測試（例如關機執行人，中斷線程），並拋出一個AssertionError。在下面的代碼中，我重新使用了超時來等待收集一定數量的數據，而不是「睡眠」。

作爲優化，使用Executor（ThreadPool）而不是創建和啓動線程。後者是相對昂貴的，使用Executor可以真正有所作爲。

在更新的代碼下方，我將它作爲應用程序運行（main方法）。（編輯28/02/17：檢查maxCollect> 0 while循環）

import java.util.*; 
import java.util.concurrent.*; 
import java.util.concurrent.atomic.AtomicBoolean; 

public class Recorder { 

    private final ExecutorService executor; 
    private Thread recordingThread; 
    private volatile boolean stopRecording; 
    private CountDownLatch finishedRecording; 
    private Collection<Object> eyeData; 
    private int maxCollect; 
    private final AtomicBoolean started = new AtomicBoolean(); 
    private final AtomicBoolean stopped = new AtomicBoolean(); 

    public Recorder() { 
     this(null); 
    } 

    public Recorder(ExecutorService executor) { 
     this.executor = executor; 
    } 

    public Recorder maxCollect(int max) { maxCollect = max; return this; } 

    private class RecordingRunnable implements Runnable { 

     @Override public void run() { 

      try { 
       int collected = 0; 
       while (!stopRecording) { 
        eyeData.add(EyeTracker.getData()); 
        if (maxCollect > 0 && ++collected >= maxCollect) { 
         stopRecording = true; 
        } 
       } 
      } finally { 
       finishedRecording.countDown(); 
      } 
     } 
    } 

    public Recorder start() { 

     if (!started.compareAndSet(false, true)) { 
      throw new IllegalStateException("already started"); 
     } 
     stopRecording = false; 
     finishedRecording = new CountDownLatch(1); 
     eyeData = new ArrayList<Object>(); 
     // the RecordingRunnable created below will see the values assigned above ('happens before relationship') 
     if (executor == null) { 
      recordingThread = new Thread(new RecordingRunnable()); 
      recordingThread.start(); 
     } else { 
      executor.execute(new RecordingRunnable()); 
     } 
     return this; 
    } 

    public Collection<Object> getData(long timeout, TimeUnit tunit) { 

     if (started.get() == false) { 
      throw new IllegalStateException("start first"); 
     } 
     if (!stopped.compareAndSet(false, true)) { 
      throw new IllegalStateException("data already fetched"); 
     } 
     if (maxCollect <= 0) { 
      stopRecording = true; 
     } 
     boolean recordingStopped = false; 
     try { 
      // this establishes a 'happens before relationship' 
      // all updates to eyeData are now visible in this thread. 
      recordingStopped = finishedRecording.await(timeout, tunit); 
     } catch(InterruptedException e) { 
      throw new RuntimeException("interrupted", e); 
     } finally { 
      stopRecording = true; 
     } 
     // if recording did not stop, do not return the eyeData (could stil be modified by recording-runnable). 
     if (!recordingStopped) { 
      throw new RuntimeException("recording"); 
     } 
     // only when everything is OK this recorder instance can be re-used 
     started.set(false); 
     stopped.set(false); 
     return eyeData; 
    } 

    public static class EyeTracker { 

     public static Object getData() { 
      try { Thread.sleep(1); } catch (Exception ignored) {} 
      return new Object(); 
     } 
    } 

    public static void main(String[] args) { 

     System.out.println("Starting."); 
     ExecutorService exe = Executors.newSingleThreadExecutor(); 
     try { 
      Recorder r = new Recorder(exe).maxCollect(50).start(); 
      int dsize = r.getData(2000, TimeUnit.MILLISECONDS).size(); 
      System.out.println("Collected " + dsize); 
      r.maxCollect(100).start(); 
      dsize = r.getData(2000, TimeUnit.MILLISECONDS).size(); 
      System.out.println("Collected " + dsize); 
      r.maxCollect(0).start(); 
      Thread.sleep(100); 
      dsize = r.getData(2000, TimeUnit.MILLISECONDS).size(); 
      System.out.println("Collected " + dsize); 
     } catch (Exception e) { 
      e.printStackTrace(); 
     } finally { 
      exe.shutdownNow(); 
      System.out.println("Done."); 
     } 
    } 
} 

編碼愉快:)