多線程編程之兩階段終止模式

       對於多線程編程，如何優雅的終止子線程，始終是一個值得考究的問題。若是直接終止線程，可能會產生三個問題：

• 子線程當前執行的任務可能必需要原子的執行，即其要麼成功執行，要麼就不執行；
• 當前任務隊列中還有未執行完的任務，直接終止線程可能致使這些任務被丟棄；
• 當前線程佔用了某些外部資源，好比打開了某個文件，或者使用了某個Socket對象，這些都是沒法被垃圾回收的對象，必須由調用方進行清理。

       因而可知，如何優雅的終止一個線程，並非一個簡單的問題。常見的終止線程的方式是，聲明一個標誌位，若是調用方要終止其建立的線程的執行，就將該標誌位設置爲須要終止狀態，子線程每次執行任務以前會檢查該標誌位，若是爲須要終止狀態，就不繼續執行任務，而是進行當前線程所佔用資源的一些清理工做，如關閉Socket和備份當前未完成的任務，清理完成以後結束當前線程的調用。

       兩階段終止模式使用的就是上述方式進行多線程的終止的，只不過其將線程的終止封裝爲一個特定的框架，使用者只須要關注特定的任務執行方式便可，從而實現了線程的終止與任務的執行的關注點的分離。兩階段終止模式的UML圖以下：

       其各角色的做用以下：

• ThreadOwner：客戶端程序，由其建立線程並執行任務，Terminatable提供的終止方法也是由其調用；
• Terminatable：終止方法提供的一個抽象接口，提供了一個terminate()方法供外部調用；
• TerminatableSupport：實現了Terminatable接口的抽象類，封裝了具體的終止模板，其doRun()是一個抽象方法，子類必須實現，用於編寫相關的任務的代碼，doTermiate()和doCleanup()方法都是鉤子方法，提供了空的實現，子類根據具體狀況判斷是否須要實現該方法；
• ConcreteTerminatable：用戶具體的終止類，其doRun()方法用於實現具體的任務；
• TerminationToken：包含了一個標誌位，而且記錄了當前線程還須要執行的任務數量，默認狀況下，只有其標誌位爲true，而且剩餘須要執行的任務數爲0時纔會真正的終止當前線程的執行。

       以下是兩階段終止模式各個類的實現，咱們首先看看Terminatable接口及其抽象實現TerminatableSupport：

public interface Terminatable {
  void terminate();
}

public abstract class TerminatableSupport extends Thread implements Terminatable {
  public final TerminationToken terminationToken;  // 記錄當前的標誌位

  public TerminatableSupport() {
    this(new TerminationToken());  // 初始化當前標誌位
  }

  public TerminatableSupport(TerminationToken terminationToken) {
    super();
    this.terminationToken = terminationToken;  // 初始化標誌位
    terminationToken.register(this);  // 註冊當前對象的標誌位
  }

  protected abstract void doRun() throws Exception;  // 供子類實現具體任務的方法

  // 鉤子方法，用於子類進行一些清理工做
  protected void doCleanup(Exception cause) {}

  // 鉤子方法，用於子類進行終止時的一些定製化操做
  protected void doTerminate() {}

  @Override
  public void run() {
    Exception ex = null;
    try {
      // 在當前線程中執行任務時，會判斷是否標識爲終止，而且剩餘任務數小於等於0，是纔會真正終止當前線程
      while (!terminationToken.isToShutdown() || terminationToken.reservations.get() > 0) {
        doRun();
      }
    } catch (Exception e) {
      ex = e;
    } finally {
      try {
        doCleanup(ex);  // 當前線程終止後須要執行的操做
      } finally {
        terminationToken.notifyThreadTermination(this);
      }
    }
  }

  @Override
  public void interrupt() {
    terminate();
  }

  @Override
  public void terminate() {
    terminationToken.setToShutdown(true);  // 設置終止狀態
    try {
      doTerminate();  // 執行客戶端定製的終止操做
    } finally {
      if (terminationToken.reservations.get() <= 0) {
        super.interrupt();  // 若是當前線程處於終止狀態，則強制終止當前線程
      }
    }
  }

  // 提供給客戶端調用的，即客戶端線程必須等待終止完成以後纔會繼續往下執行
  public void terminate(boolean waitUntilThreadTerminated) {
    terminate();
    if (waitUntilThreadTerminated) {
      try {
        this.join();
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
  }
}

       當客戶端調用termiante()方法時，其首先會將當前的終止狀態設置爲true，而後調用doTerminate()方法，這裏須要注意的一點是，若是當前線程在doRun()方法中處於等待狀態，好比Thread.sleep()、Thread.wait()方法等，那麼即便設置了終止狀態，也沒法使其被喚醒，由於其沒法運行到檢測終止狀態的代碼處，其只能使用intertupt()方法才能使其被喚醒並終止，可是對於Socket.read()方法，即便調用了interrupt()方法，也沒法使其終止，於是這裏設置了doTerminate()方法，用於子類在該方法中關閉Socket。最後在finally塊中，調用super.interrupt()方法，該調用的做用也即若是當前線程在doRun()方法中被阻塞，就強制終止其執行。

public class TerminationToken {
  protected volatile boolean toShutdown = false;  // 終止狀態的標誌位
  public final AtomicInteger reservations = new AtomicInteger(0);  // 記錄當前剩餘任務數

  // 記錄了全部註冊了TerminationToken的實例，這裏使用Queue是由於可能會有多個
  // Terminatable實例共享同一個TeraminationToken，若是是共享的，那麼reservations
  // 實例就保存了全部共享當前TerminationToken實例的線程所須要執行的任務總數
  private final Queue<WeakReference<Terminatable>> coordinatedThreads;

  public TerminationToken() {
    coordinatedThreads = new ConcurrentLinkedQueue<>();
  }

  public boolean isToShutdown() {
    return toShutdown;
  }

  public void setToShutdown(boolean toShutdown) {
    this.toShutdown = toShutdown;
  }

  // 將當前Terminatable實例註冊到當前TerminationToken中
  protected void register(Terminatable thread) {
    coordinatedThreads.add(new WeakReference<>(thread));
  }

  // 若是是多個Terminatable實例註冊到當前TerminationToken中，
  // 則廣播當前的終止狀態，使得這些實例都會終止
  protected void notifyThreadTermination(Terminatable thread) {
    WeakReference<Terminatable> wrThread;
    Terminatable otherThread;
    while (null != (wrThread = coordinatedThreads.poll())) {
      otherThread = wrThread.get();
      if (null != otherThread && otherThread != thread) {
        otherThread.terminate();
      }
    }
  }
}

       關於Terminatable和TerminationToken的關係是一對多的關係，即多個Terminatable實例可共用一個TerminationToken實例，而其reservations屬性所保存的則是這多個Terminatable實例所共同要完成的任務數量。這裏典型的多個Terminatable共用一個TerminationToken實例的例子是當有多個工做者線程時，這幾個線程所消費的任務是共用的，於是其TermiantionToken實例也須要共用。

       兩階段終止模式的使用場景很是的多，基本上只要是使用了子線程的位置都須要使用必定的方式來優雅的終止該線程的執行。咱們這裏使用生產者和消費者的例子來演示兩階段終止模式的使用，以下是該例子的代碼：

public class SomeService {
  private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);

  private final Producer producer = new Producer();
  private final Consumer consumer = new Consumer();

  public static void main(String[] args) throws InterruptedException {
    SomeService ss = new SomeService();
    ss.init();
    TimeUnit.SECONDS.sleep(500);
    ss.shutdown();
  }

  // 中止生產者和消費者的執行
  public void shutdown() {
    producer.terminate(true);  // 先中止生產者，只有在生產者徹底中止以後纔會中止消費者
    consumer.terminate();  // 中止消費者
  }

  // 啓動生產者和消費者
  public void init() {
    producer.start();
    consumer.start();
  }

  // 生產者
  private class Producer extends TerminatableSupport {
    private int i = 0;

    @Override
    protected void doRun() throws Exception {
      queue.put(String.valueOf(i++));  // 將任務添加到任務隊列中
      consumer.terminationToken.reservations.incrementAndGet();  // 更新須要執行的任務數量
    }
  }

  // 消費者
  private class Consumer extends TerminatableSupport {
    @Override
    protected void doRun() throws Exception {
      String product = queue.take();  // 獲取任務
      System.out.println("Processing product: " + product);
      try {
        TimeUnit.SECONDS.sleep(new Random().nextInt(100));  // 模擬消費者對任務的執行
      } catch (InterruptedException e) {
        // ignore
      } finally {
        terminationToken.reservations.decrementAndGet();  // 更新須要執行的任務數量
      }
    }
  }
}

       能夠看到，在子類使用兩階段終止模式時，其只須要實現各自所須要執行的任務，而且更新當前任務的數量便可。在某些狀況下，當前任務的數量也能夠不進行更新，好比在進行終止時，不關心當前剩餘多少任務須要執行。