Java並發編程系列之二十九：正確終止與恢復線程（續）

而其他代碼保持不變，再次運行以上的程序，發現能夠正確終止了。主要就是使用中斷機制完成了線程之間的協作，從而達到正確終止線程的目的。

實際上在調用可阻塞的方法時拋出的InterruptedException是為了讓調用者能夠注意到中斷信息，使得調用者可以就中斷做出自己的操作。往往在將中斷信息傳給調用者之前需要執行其他操作，如果在線程中使用中斷機制完成線程之間的協作，那麼就應該調用Thread.currentThread().intrrupt()恢復當前線程的中斷狀態，這樣當前線程就能夠繼續其他操作了。正常情況下，都需要對中斷進行響應，除非自己實現了中斷所應該進行的操作。

為了取消線程的執行，除了之前的方法，還可以使用Future.get(Long time,TimeUnit unit)的帶超時限制的方法取消線程的執行，如果沒有在指定的時間內完成任務，那麼可以在代碼中直接調用Future.cancel()方法取消任務的執行。取消任務的時候有兩種情況：一是任務在指定的時間完成了，這個時候調用取消操作沒有什麼影響；二是任務沒有在指定的時間完成，那麼調用cancel方法後任務將被中斷。

偽代碼如下：

Future task = threadPool.submit(runnable);
try{

}catch(TimeOutException e){
    //會取消任務的執行
}catch(ExecutionException e){
    //如果在任務中拋出了執行異常，則重新拋出該異常
    throw(new Throwable(e.getCause()));
}finally{
    //true表示正在執行的任務能夠接收中斷，如果在執行則線程能被中斷
    //如果為false，則表示若任務還沒有啟動則不要啟動該任務
    task.cancel(true);
}

實現線程取消的完整例子

這裡以日志服務作為例子，業務場景是這樣的：前台會有多個生產者調用日志服務輸出程序的日志，生產者將需要輸出的日志信息放入一個隊列中，後台服務器有一個消費者線程，負責從隊列中取出日志信息並輸出（目的地可能不同）。顯然這是一個典型的生產者-消費者問題，不過這裡出現了多個生產者，但是只有一個消費者。顯然如果生產者的速度遠遠大於消費者的處理速度的話，很可能造成阻塞，不過這點已經再上面的分析中得到了解決。現在需要實現的是，提供可靠的關閉日志服務的方法，在前台調用服務接口可以正確停止日志服務，而不會出現任何問題。

實現代碼如下：

package com.rhwayfun.patchwork.concurrency.r0411;

import java.io.PrintWriter;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by rhwayfun on 16-4-11.
 */
public class LoggerService {
    // 存放日志消息的阻塞隊列
    private final BlockingQueue logQueue;
    // 打印日志的消費者線程
    private final LoggerThread loggerThread;
    // 打印日志的打印器
    private PrintWriter writer;
    // 日志服務是否關閉的標志
    private boolean isShutdown;
    // 執行log方法的調用者的計數器
    private int reservations;

    public LoggerService(PrintWriter writer) {
        this.logQueue = new LinkedBlockingQueue<>(5);
        this.loggerThread = new LoggerThread(writer);
    }

    /**
     * 啟動日志服務
     */
    public void start() {
        loggerThread.start();
    }

    /**
     * 記錄日志
     *
     * @param msg
     * @throws InterruptedException
     */
    public void recordLog(String msg) throws InterruptedException {
        // 有條件保持對日志的添加
        // 並且在接收到關閉請求時停止往隊列中填入日志
        synchronized (this) {
            if (isShutdown) throw new IllegalStateException("LoggerService is shutdown!");
            ++reservations;
        }
        // 由生產者將消息放入隊列
        // 這裡不放入synchronized塊是因為put方法有阻塞的作用
        logQueue.put(msg);
    }

    /**
     * 停止日志服務
     */
    public void stop() {
        // 以原子方式檢查關閉請求
        synchronized (this) {
            isShutdown = true;
        }
        // 讓消費者線程停止從隊列取日志
        loggerThread.interrupt();
    }

    /**
     * 消費者線程
     */
    private class LoggerThread extends Thread {
        private PrintWriter writer;

        public LoggerThread(PrintWriter writer) {
            this.writer = writer;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    try {
                        // 持有的鎖與之前的相同
                        // 如果接收到應用程序的關閉請求並且沒有生產者線程繼續往隊列填入日志
                        // 那麼就結束循環，消費者線程終止
                        synchronized (LoggerService.this) {
                            if (isShutdown && reservations == 0) break;
                        }
                        // 從隊列獲取生產者的日志
                        String msg = logQueue.take();
                        // 每輸出一條日志就減少一個線程
                        synchronized (LoggerService.this) {
                            --reservations;
                        }
                        writer.println("Read: " + msg);
                    } catch (InterruptedException e) {
                        //恢復中斷狀態
                        Thread.currentThread().interrupt();
                    }
                }
            } finally {
              writer.close();
            }
        }
    }

    /**
     * 生產者線程
     */
    private static class LoggerWriter implements Runnable {
        private LoggerService service;
        private final DateFormat format = new SimpleDateFormat("HH:mm:ss");

        public LoggerWriter(LoggerService service) {
            this.service = service;
        }

        @Override
        public void run() {
            try {
                String msg = "time is " + format.format(new Date());
                System.out.println("Write: " + msg);
                service.recordLog(msg);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {

        LoggerService service = new LoggerService(new PrintWriter(System.out));
        //創建多個生產者線程負責創建日志
        for (int i = 0; i < 5; i++) {
            new Thread(new LoggerWriter(service)).start();
            TimeUnit.SECONDS.sleep(1);
        }
        //啟動日志服務
        service.start();
        //休眠10秒
        TimeUnit.SECONDS.sleep(10);
        //關閉日志服務
        service.stop();
    }
}