[本文是我對Java Concurrency In Practice 7.1的歸納和總結(jié). ?轉(zhuǎn)載請注明作者和出處, ?如有謬誤, 歡迎在評論中指正. ]
啟動線程之后, 大多數(shù)時候我們等待線程運行完成后自動結(jié)束. 但是有時我們希望可以提前終止線程的運行:
1. 用戶申請取消時. 比如用戶點擊了取消按鈕.
2. 時間限制的任務. 有些任務具有時間限制, 如果在一定的時間內(nèi)仍然沒有得到想要的結(jié)果, 我們可能希望終止該任務的運行.
3. 發(fā)生特定的事件時. 比如多個線程同時在不同的位置搜索某一文件, 當其中一個線程搜索到了想要的文件, 應該終止其他仍在運行的線程.
4. 發(fā)生錯誤時. 比如發(fā)生了磁盤已滿的錯誤, 需要向磁盤寫入數(shù)據(jù)的線程應該提前終止.
5. 應用或者服務被關閉時.
?
java沒有直接規(guī)定如何安全的提前終止線程的運行, 相反, 提供了不具約束力的協(xié)商式機制: 線程A可以請求線程B中斷, 但是是否響應, 何時響應, 如何響應中斷請求, 由線程B自己決定. 每個線程對象都有一個boolean型的中斷標記, 其他線程請求目標線程中斷時, 會將目標線程的中斷標記設置為true, 然后由目標線程自己決定如何處理. 所以中斷線程時, 我們需要知道目標線程的中斷機制. 如果我們不知道目標線程會怎樣處理中斷請求, 不要貿(mào)然請求其中斷. Thread類中與中斷標記相關的方法有:
public class Thread {
// 請求線程中斷, 該方法會將線程的中斷標記設置為true. 如何處理中斷由目標線程決定
public void interrupt() { ... }
// 返回中斷標記的值
public boolean isInterrupted() { ... }
// 這個方法的命名很讓人蛋疼. 該靜態(tài)方法用于重置中斷標記(將其設置為false), 并返回重置之前的值
public static boolean interrupted() { ... }
...
}
?
設置自定義flag結(jié)束線程
在深入了解java的中斷機制之前, 我們先看一個通過設置自定義的flag結(jié)束線程的例子:
public class PrimeGenerator implements Runnable {
private final List<BigInteger> primes = new ArrayList<BigInteger>();
/**
* 自定義的flag, 為保證線程可見性, 將其聲明為volatile
*/
private volatile boolean cancelled;
public void run() {
BigInteger p = BigInteger.ONE;
// 每次循環(huán)之前檢查cancelled標記的值, 如果cancelled為true, 循環(huán)終止, 線程也就運行結(jié)束了
while (!cancelled) {
p = p.nextProbablePrime();
synchronized (this) {
primes.add(p);
}
}
}
public void cancel() {
cancelled = true;
}
public synchronized List<BigInteger> get() {
return new ArrayList<BigInteger>(primes);
}
public static void main(String[] args) {
PrimeGenerator generator = new PrimeGenerator();
Thread t = new Thread(generator);
t.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 通過調(diào)用cancel方法, 將自定義的cancelled標記設置為true, 從而使得線程t運行終止
generator.cancel();
System.out.println(generator.get().size());
}
}?
?
自定義flag結(jié)束線程存在的問題
PrimeGenerator自定義了cancelled標記, 在繼續(xù)下一次循環(huán)之前, 輪詢該標記的值. 當cancelled標記為true時, 循環(huán)不再繼續(xù).
這種方式在PrimeGenerator中可以起到期望的作用, 但使用這種方式結(jié)束線程存在潛在的問題: 假如循環(huán)中執(zhí)行了阻塞操作, 那么即使cancelled標記被設置為true, run方法卻沒有機會去檢查cancelled標記的值, 所以線程將遲遲無法結(jié)束:
class BrokenPrimeProducer extends Thread {
private final BlockingQueue<BigInteger> queue;
private volatile boolean cancelled = false;
BrokenPrimeProducer(BlockingQueue<BigInteger> queue) {
this.queue = queue;
}
public void run() {
try {
BigInteger p = BigInteger.ONE;
while (!cancelled) {
// 當隊列已滿時, put方法將會阻塞. 一旦put方法阻塞, 且沒有其他線程從隊列中取數(shù)據(jù)時, 阻塞將一直持續(xù)下去
queue.put(p = p.nextProbablePrime());
}
} catch (InterruptedException consumed) {
//...
}
}
public void cancel() {
cancelled = true;
}
public static void main(String[] args) {
// 設置隊列的最大容量為10
BlockingQueue<BigInteger> primes = new LinkedBlockingQueue<BigInteger>(10);
BrokenPrimeProducer producer = new BrokenPrimeProducer(primes);
producer.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
producer.cancel();
}
}
在主線程中啟動BrokenPrimeProducer線程, 1s后調(diào)用其cancel方法. 隊列的最大容量被設定為10, 1s后隊列肯定已滿, 也就是說BrokenPrimeProducer線程將在put方法上阻塞, 沒有機會去檢查cancelled標記, 從而導致BrokenPrimeProducer線程無法結(jié)束.
?
可中斷的阻塞方法
java API中的大多數(shù)阻塞方法都是可中斷的, 如Thread.sleep, Object.wait, BlockingQueue.put等. 可中斷的阻塞方法有一個共同的特點: 聲明拋出InterruptedException異常. 可中斷的阻塞方法在阻塞期間會周期性檢查當前線程的中斷標記, 如果發(fā)現(xiàn)當前線程的中斷標記為true, 就重置中斷標記后提前從阻塞狀態(tài)返回, 并拋出InterruptedException異常.?
據(jù)此我們可以改進BrokenPrimeProducer類:
class PrimeProducer extends Thread {
private final BlockingQueue<BigInteger> queue;
PrimeProducer(BlockingQueue<BigInteger> queue) {
this.queue = queue;
}
public void run() {
try {
BigInteger p = BigInteger.ONE;
// 每次循環(huán)前檢查當前線程的中斷標記, 如果中斷標記為設定為true, 則循環(huán)結(jié)束
// 就算當前線程阻塞在put方法上, 在阻塞期間也會周期性檢查中斷標記, 一旦發(fā)現(xiàn)中斷標記為true, 就會從阻塞狀態(tài)中返回, 并拋出InterruptedException異常
while (!Thread.currentThread().isInterrupted()) {
queue.put(p = p.nextProbablePrime());
}
} catch (InterruptedException consumed) {
System.out.println("InterruptedException happened");
}
}
public void cancel() {
// interrupt方法會將當前線程的中斷標記設置為true
interrupt();
}
public static void main(String[] args) {
// 設置隊列的最大容量為10
BlockingQueue<BigInteger> primes = new LinkedBlockingQueue<BigInteger>(10);
PrimeProducer producer = new PrimeProducer(primes);
producer.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
producer.cancel();
}
}
?
通過Future終止線程運行
有時我們將task提交給線程池運行, 由于我們不知道task會由線程池中的哪一個線程運行, 也不知道線程池中的線程會怎樣處理中斷, 所以無法直接調(diào)用Thread對象的interrupt方法提前終止線程的運行. 但是ExecutorService類的submit, invokeAll等方法會返回表示task未決結(jié)果的Future對象, 調(diào)用Future對象的cancel方法, 可以取消task的運行. Future類中與取消有關的方法有:
1. boolean cancel(boolean mayInterruptIfRunning). 該方法嘗試取消task的執(zhí)行. 如果task已經(jīng)完成, 或已取消, 或由于某些原因無法取消, 則嘗試失敗, 返回false.?
如果task尚未啟動, 則成功調(diào)用其Future對象的cancel方法將導致其永不啟動.?
mayInterruptIfRunning如果為true, 且此時task正在某個線程中運行, 那么該線程的中斷標記將被設置為true.?
當mayInterruptIfRunning為false時, 如果task沒有啟動則不再啟動, 如果task已經(jīng)啟動, 則嘗試失敗.?
如果task沒有處理中斷, mayInterruptIfRunning應該為false.?
此方法返回后, isDone方法將始終返回true. 如果此方法返回true, 對isCancelled方法的后續(xù)調(diào)用將始終返回true.
2. boolean isDone(). 如果task已經(jīng)完成, 該方法返回true. 完成的情況包括正常完成, task被取消, 異常終止等.
3. boolean isCancelled(). 如果task正常完成前被取消, 該方法返回true.
前面提到, 如果不知道線程會怎樣處理中斷, 就不應該調(diào)用該線程的interrupt方法, 那么調(diào)用Future的cancel方法, 并將mayInterruptIfRunning參數(shù)設置為true是否合適? 線程池中用于執(zhí)行task的線程會將中斷的處理委托給task, 所以這樣做是合適的(前提是task正確處理了中斷).
使用Future取消task的例子:
/**
* 執(zhí)行一項任務, 如果指定時間內(nèi)沒有正常完成, 就取消該任務
*/
public static void timedRun(Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
Future<?> task = taskExec.submit(r);
try {
// 如果線程池中的線程執(zhí)行任務過程中該線程發(fā)生了中斷, 那么調(diào)用task的get方法將會拋出InterruptedException異常.
// 對于InterruptedException, 按照之前總結(jié)的方法處理即可. 此例將其拋給上層.
task.get(timeout, unit);
} catch (TimeoutException e) {
// 如果發(fā)生TimeoutException異常, 表明執(zhí)行時間超時, 此時取消該任務即可
} catch (ExecutionException e) {
// 發(fā)生其他異常時, 不僅要取消任務的執(zhí)行, 也應該重拋該異常
throw launderThrowable(e.getCause());
} finally {
task.cancel(true);
}
}
?
線程的中斷方式總結(jié):
1. 可以通過設置自定義標記結(jié)束線程. 但是這樣方式在包含阻塞方法的任務中不適用.
2. interrupt線程. 前提是知道目標線程會怎樣處理interrupt請求.
3. 如果是提交給線程池運行的任務, 可以調(diào)用Future.cancel.
?