資訊專欄INFORMATION COLUMN
摘要：某些編程語言被設計為可以將并發任務彼此隔離，這些語言通常被稱為函數性語言。通過使用多線程機制，這些獨立任務也被稱為子任務中的每一個都將由執行線程來驅動。
并發
之前學的都是順序編程的知識，學習并發編程就好像進入了一個全新的領域，有點類似于學習了一門新的編程語言，或者至少是學習了一整套新的語言概念。要理解并發編程，其難度與理解面向對象編程差不多。如果花點兒功夫，就能明白其基本機制，但想要抓住其本質，就需要深入的學習和理解。所以看完《Java編程思想》或許會變得過分自信，但寫復雜的多線程時，應該多看其他多線程的書籍，關鍵還是多動手。
“并發是一種具有可論證的確定性，但實際上具有不可確定性。”
使用并發時，你的自食其力，并且只有變得多疑而自信，才能用Java編寫出可靠的多線程代碼。
用并發解決的問題大體可以分為“速度”和“設計可管理性”兩種。 速度并發解決“速度”問題不僅僅是利用多個CPU去解決分片的問題，也就是說并發不僅僅是多個CPU的事情，也是單個CPU的事情。如果提高程序在單個CPU的性能，就得考慮具體情況，正常情況單個CPU運行多任務（task）是有上下文切換的性能損耗。但在阻塞（Blocking）的情況下就不同了。
我們先看看阻塞的定義：如果程序中的某個任務因為該程序控制范圍之外的某些條件（通常是I/O），那我們就說這個任務或線程阻塞了。
如果使用并發來寫這個阻塞程序，在一個任務阻塞時，程序中的其他任務還可以繼續執行。這樣性能會有很大的提升。所以如果沒有阻塞的情況，在單CPU使用并發，就沒必要了。
在單個CPU的系統中性能提高的常見示例:事件驅動編程（event-driven programing）。
實現并發最直接的方式是在操作系統級別使用進程（process）。多任務操作系統可以通過周期性地將CPU從一個進程切換到另一個進程，來實現同時運行多個進程（程序）。
某些編程語言被設計為可以將并發任務彼此隔離，這些語言通常被稱為函數性語言。Erlang就是這樣的語言，它包含針對任務之間彼此通信的安全機制。如果你發現程序中某個部分必須大量使用并發，并且你在試圖構建這個部分時遇到過多的問題。那么你可以考慮使用像Erlang這類專門的并發語言來創建這個部分。
Java語言采用更加傳統的方式，在順序語言的基礎上提供對線程的支持。 Java的目的是“編寫一次，到處運行”，所以在OSX之前的Macintosh操作系統版本是不支持多任務，因此Java支持多線程機制，讓并發Java程序能夠移植到Macintosh和類似的平臺上。
設計可管理性設計可管理性，我更愿意說是一個解決問題的方法模型（程序設計）。線程使你能夠創建更加松散耦合的設計。
在單CPU上使用多任務的程序（代碼）在任意時刻仍然只能執行一項任務，因此理論上講，肯定可以不用任何任務就可以編寫相同的程序。但是，這樣寫來的代碼可能會很混亂，不方便維護。因此并發提供一種重要的組織結構上的好處：你的程序設計可以極大地簡化。某些類似的問題，例如仿真，沒有并發的支持是很難解決的。
搶占式調度指的是每條線程執行的時間、線程的切換都是由系統控制，每條線程可能都分同樣的的執行時間片（CPU切片），也可能是在某些線程執行的時間片較長，甚至某些線程得不到執行時間片。這種機制下，優點是一個線程阻塞不會導致整個進程堵塞，缺點就是上下文切換開銷大。
協同式調度指的是某一條線程執行完后主動通知系統切到另一條線程上執行。線程的執行時間由線程本身控制，線程切換可以預知。優點是不存在多線程同步問題，上下文切換開銷小，缺點是如果一個線程阻塞了，那么可能造成整個系統崩潰。
Java線程機制是搶占式.
線程讓出cpu的情況：
1.當前運行線程主動放棄CPU，JVM暫時放棄CPU操作（基于時間片輪轉調度的JVM操作系統不會讓線程永久放棄CPU，或者說放棄本次時間片的執行權），例如調用yield()方法。
2.當前運行線程因為某些原因進入阻塞狀態，例如阻塞在I/O上。
3.當前運行線程結束，即運行完run()方法里面的任務
并發需要付出代價，包含復雜性代價。但這些代價與優化程序設計、資源負載均衡以及用戶體驗上的改進相比，這些代價就顯得微不足道。
線程帶來設計上的演變為了獲取線程的結果，于是產生輪詢，然后再后來為了解決輪詢，引進了靜態方法的回調，再后來帶來實例方法的回調，最后引出設計模式：策略模式 和Java5引進多線程編程的新方法，通過隱藏細節可以更容易地處理回調——ExecutorService和Futrue
輪詢例子：
package com.jc.thread;
import com.jc.thinkinjava.io.util.Directory;
import javax.xml.bind.DatatypeConverter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.security.DigestInputStream;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
/**
* 回調例子-前序
*
* 計算文件的256位的SHA-2消息摘要
* 由于瓶頸在IO上，所以采用多線程
*
* 嘗試去獲取線程返回的值，但發現需要另外個線程不停的輪詢，這是很耗cpu資源
*/
@SuppressWarnings("Duplicates")
public class ReturnDigest extends Thread {
private String fileName;
private byte[] digest;
public ReturnDigest(String fileName) {
this.fileName = fileName;
}
@Override
public void run() {
try {
// System.out.println(fileName);
FileInputStream in = new FileInputStream(fileName);
MessageDigest sha = MessageDigest.getInstance("SHA-256");
DigestInputStream digestInputStream = new DigestInputStream(in, sha);
while (digestInputStream.read() != -1) ;
digestInputStream.close();
digest = sha.digest(); //注意，不是DigestInputStream的方法哦
StringBuilder sb = new StringBuilder(fileName);
sb.append(":").append(DatatypeConverter.printHexBinary(digest));
System.out.println(sb.toString());
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (NoSuchAlgorithmException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
public byte[] getDigest() {
return this.digest;
}
public static void main(String[] args) {
File[] files = Directory.local(".", ".*");
List fileList = new ArrayList();
for (int i = 0; i < files.length; i++) {
File file = files[i];
if (!file.isDirectory()) {
fileList.add(file);
}
}
ReturnDigest[] digests = new ReturnDigest[fileList.size()];
for (int i = 0; i < fileList.size(); i++) {
File file = fileList.get(0);
digests[i] = new ReturnDigest(file.getAbsolutePath());
digests[i].start();
}
for(int i=0;i
然后為了解決輪詢，產生了靜態方法的回調：
package com.jc.thread.callback;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.security.DigestInputStream;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
/**
* 回調例子
* 靜態方法的回調
*/
@SuppressWarnings("Duplicates")
public class CallbackDigest implements Runnable{
private String fileName;
public CallbackDigest(String fileName) {
this.fileName = fileName;
}
@Override
public void run() {
try {
// System.out.println(fileName);
FileInputStream in = new FileInputStream(fileName);
MessageDigest sha = MessageDigest.getInstance("SHA-256");
DigestInputStream digestInputStream = new DigestInputStream(in, sha);
while (digestInputStream.read() != -1) ;
digestInputStream.close();
byte[] digest = sha.digest(); //注意，不是DigestInputStream的方法哦
CallbackDigestUserInterface.receiveDigest(digest,fileName);
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (NoSuchAlgorithmException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
}
package com.jc.thread.callback;
import com.jc.thinkinjava.io.util.Directory;
import com.jc.thread.DigestRunnable;
import javax.xml.bind.DatatypeConverter;
import java.io.File;
/**
* 回調例子
* 靜態方法的回調
*/
public class CallbackDigestUserInterface {
public static void receiveDigest(byte[] digest,String fileName){
StringBuilder sb = new StringBuilder(fileName);
sb.append(":").append(DatatypeConverter.printHexBinary(digest));
System.out.println(sb.toString());
}
public static void main(String[] args) {
File[] files = Directory.local(".", ".*");
for (File file : files) {
if (!file.isDirectory())
new Thread(new DigestRunnable(file.getAbsolutePath())).start();
}
}
}
實例方法的回調：
package com.jc.thread.callback;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.security.DigestInputStream;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
public class InstanceCallbackDigest implements Runnable{
private String fileName;
private InstanceCallbackDigestUserInterface callback;
public InstanceCallbackDigest(String fileName, InstanceCallbackDigestUserInterface instanceCallbackDigestUserInterface) {
this.fileName = fileName;
this.callback = instanceCallbackDigestUserInterface;
}
@Override
public void run() {
try {
// System.out.println(fileName);
FileInputStream in = new FileInputStream(fileName);
MessageDigest sha = MessageDigest.getInstance("SHA-256");
DigestInputStream digestInputStream = new DigestInputStream(in, sha);
while (digestInputStream.read() != -1) ;
digestInputStream.close();
byte[] digest = sha.digest(); //注意，不是DigestInputStream的方法哦
callback.receiveDigest(digest);
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (NoSuchAlgorithmException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
}
package com.jc.thread.callback;
import com.jc.thinkinjava.io.util.Directory;
import com.jc.thread.ReturnDigest;
import javax.xml.bind.DatatypeConverter;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
/**
* 回調例子
*
* 使用實例方法代替靜態方法進行回調
*
* 雖然復雜點，但優點很多。如：
* 1. 主類（InstanceCallbackDigestUserInterface）的各個實例映射為一個文件，可以很自然地記錄跟蹤這個文件的信息，而不需要額外的數據結構
* 2. 這個實例在有必要時可以容易地重新計算某個特定文件的摘要
*
* 實際上，經證明，這種機制有更大的靈活性。
*
* 這種機制，也稱為：觀察者模式，如Swing、AWT
*/
public class InstanceCallbackDigestUserInterface {
private String fileName;
private byte[] digest;
public InstanceCallbackDigestUserInterface(String fileName) {
this.fileName = fileName;
}
public void calculateDigest() {
InstanceCallbackDigest instanceCallbackDigest = new InstanceCallbackDigest(fileName, this);
new Thread(instanceCallbackDigest).start();
}
public void receiveDigest(byte[] digest) {
this.digest = digest;
System.out.println(this);
}
@Override
public String toString() {
String result = fileName + ": ";
if (digest != null) {
result += DatatypeConverter.printHexBinary(digest);
} else {
result += "digest not available";
}
return result;
}
public static void main(String[] args) {
File[] files = Directory.local(".", ".*");
List fileList = new ArrayList();
for (int i = 0; i < files.length; i++) {
File file = files[i];
if (!file.isDirectory()) {
fileList.add(file);
}
}
for (int i = 0; i < fileList.size(); i++) {
File file = fileList.get(0);
InstanceCallbackDigestUserInterface instanceCallbackDigestUserInterface = new InstanceCallbackDigestUserInterface(file.getAbsolutePath());
instanceCallbackDigestUserInterface.calculateDigest();
}
}
}
Java5引進的新方法，ExecutorService和Future：
package com.jc.thread.callback;
import java.util.concurrent.Callable;
public class FindMaxTask implements Callable {
private int[] data;
private int start;
private int end;
public FindMaxTask(int[] data, int start, int end) {
this.data = data;
this.start = start;
this.end = end;
}
@Override
public Integer call() throws Exception {
int max = Integer.MAX_VALUE;
for (int i = start; i < end; i++) {
if (data[i] > max) max = data[i];
}
return max;
}
}
package com.jc.thread.callback;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
*
* Java5引入了多線程編程的一個新方法，通過隱藏細節可以更容易地處理回調
* 使用回調實現的Futrue
*/
public class MultithreadedMaxFinder {
public static int max(int[] data) throws ExecutionException, InterruptedException {
if (data.length == 1) {
return data[0];
} else if (data.length == 0) {
throw new IllegalArgumentException();
}
FindMaxTask task1 = new FindMaxTask(data,0,data.length/2);
FindMaxTask task2 = new FindMaxTask(data,data.length/2,data.length);
ExecutorService executorService = Executors.newFixedThreadPool(2);
Future future1 = executorService.submit(task1);
Future future2 = executorService.submit(task2);
//調用future1.get()時，這個方法會進行阻塞，等待第一個FindMaxTask完成。只有當第一個FindMaxTask完成，才會調用future2.get()
return Math.max(future1.get(),future2.get());
}
}
基本線程機制
并發編程使我們可以將程序劃分為多個分離的、獨立運行的任務。通過使用多線程機制，這些獨立任務（也被稱為子任務）中的每一個都將由執行線程來驅動。
線程模型：一個線程就是進程中的一個單一順序控制流，因此單個進程可以擁有多個并發執行的任務，感覺每個任務都好像有其CPU一樣，其底層機制是切分CPU時間，但通常不用考慮CPU的切片。
線程模型為編程帶來便利，它簡化了在單一程序中同時交織在一起的多個操作的處理。在使用線程時，CPU將輪流給每個任務分配其占用時間。線程的一大好處是可以使你從這一個層次抽身出來，即代碼不必知道它是運行在具有一個還是多個CPU的機子上。
所以，使用線程機制是一種建立透明的、可擴展的程序的方法，如果程序運行得太慢，為機器增添一個CPU就能容易地加快程序的運行速度。多任務和多線程往往是使用多處理器系統的最合理方式。
//此方法調用是對 線程調度器 的一種建議：我已經執行完生命周期中最重要的部分了，此刻正是切換給其他任務執行一段時間的大好時機。
Thread.yield();
Thread.yield();這個方法叫“讓步”，不過沒有任何機制保證它將會被采納。
術語
在Java中學習并發編程，總是會讓人困惑。讓人困惑是那些概念，特別是涉及到線程。
要執行的任務和驅動它的線程，這里的任務和線程是不同的，在Java中會更明細，因為你對Thread類實際沒有任何控制權（特別是使用Executor時候）。通過某種方式，將任務附著到線程，以使這個線程可以驅動任務。
在Java中，Thread類自身不執行任何操作，它只是驅動賦予它的任務，但是線程的一些研究中，總是使用這樣的話語“線程執行這項或那項動作”，仿佛“線程就是任務”。這一點是讓新人是十分困惑的。因為會讓人覺得任務和線程是一種“是一個”的關系。覺得應該從Thread繼承出一個任務。但實際不是，所以用Task名字會更好。
那為什么Java設計者不用Task而用Thread或Runnable呢？ 之所以有上述的困惑（概念混淆），那是因為，雖然從概念上講，我們應該只關注任務，而不需要關注線程的細節，我們只需要定義任務，然后說“開始”就好。但實際情況是，在物理上，創建線程可能會代價很高，因此需要人工去保存和管理它們。而且Java的線程機制是基于C的低級的P線程（pthread）方式。所以才導致任務和線程這兩個概念總是混在一起。站在實現和更抽象的角度，這兩者應該分開，所以編寫代碼時，你必須遵守規則。
為了描述更清楚，因為定義為要執行的工作則為“任務”，引用到驅動任務的具體機制時，用“線程”。 如果只是概念級別上討論系統，則只用“任務”就行。
加入一個線程
一個線程可以調用其他線程的join()方法，其效果是等待一段時間直到第二個線程結束才繼續執行。
package com.jc.concurrency;
/**
* 一個線程可以等待一個線程完成，那就是用join
* @author
*
*/
class Sleeper extends Thread {
private int duration;
public Sleeper(String name, int sleepTime) {
super(name);
duration = sleepTime;
start();
}
public void run() {
try {
sleep(duration);
} catch (InterruptedException e) { //異常捕獲時會將Interrupted這個標志位重置為false，所以在這里輸出false
System.out.println(getName() + " was interrupted. " + "isInterrupted(): " + isInterrupted());
return;
}
System.out.println(getName() + " has awakened");
}
}
class Joiner extends Thread {
private Sleeper sleeper;
public Joiner(String name, Sleeper sleeper) {
super(name);
this.sleeper = sleeper;
start();
}
public void run() {
try {
sleeper.join();
} catch (InterruptedException e) {
System.out.println("Interrupted");
}
System.out.println(getName() + " join completed");
}
}
public class Joining {
public static void main(String[] args) {
Sleeper sleepy = new Sleeper("Sleepy", 1500), grumpy = new Sleeper("Grumpy", 1500);
Joiner dopey = new Joiner("Dopey", sleepy), doc = new Joiner("Doc", grumpy);
grumpy.interrupt();
}
}
捕獲異常
在main方法是無法捕獲到線程里的異常。為解決這個問題，我們修改Executor產生線程的方式。Java SE5中的新接口：Thread.UncaughtExceptionHandler
package com.jc.concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
* 使用Thread.UncaughtExceptionHandler處理線程拋出的異常
*
* MyUncaughtExceptionHandler會新建線程去處理其他線程跑出來的異常
*
* @author
*
*/
class ExceptionThread2 implements Runnable {
public void run() {
Thread t = Thread.currentThread();
System.out.println("run() by " + t);
System.out.println("eh = " + t.getUncaughtExceptionHandler());
throw new RuntimeException();
}
}
class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
public void uncaughtException(Thread t, Throwable e) {
System.out.println("caught " + t + ""s " + e);
}
}
class HandlerThreadFactory implements ThreadFactory {
@Override
public Thread newThread(Runnable r) {
System.out.println(this + " creating new Thread");
Thread t = new Thread(r);
System.out.println("created " + t);
t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
System.out.println("eh = " + t.getUncaughtExceptionHandler());
return t;
}
}
public class CaptureUncaughtException {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool(new HandlerThreadFactory());
exec.execute(new ExceptionThread2());
}
}/*
* output:
*
* com.jc.concurrency.HandlerThreadFactory@4e25154f creating new Thread
* created Thread[Thread-0,5,main] eh =
* com.jc.concurrency.MyUncaughtExceptionHandler@70dea4e run() by
* Thread[Thread-0,5,main] eh =
* com.jc.concurrency.MyUncaughtExceptionHandler@70dea4e
* com.jc.concurrency.HandlerThreadFactory@4e25154f creating new Thread
* created Thread[Thread-1,5,main] eh =
* com.jc.concurrency.MyUncaughtExceptionHandler@5490c2f5 caught
* Thread[Thread-0,5,main]"s java.lang.RuntimeException
*
*
*
*/
還可以設置默認異常處理器：
package com.jc.concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 設置默認的線程異常處理類
* @author
*
*/
public class SettingDefaultHandler {
public static void main(String[] args) {
Thread.setDefaultUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new ExceptionThread());
}
}
線程狀態（Thread state）1.新建(new)：
一個線程可以處于四種狀態之一：新建(new)，就緒(Runnable)，阻塞(Blocked)，死亡(Dead)。
1.新建(new)：這是個短暫狀態，當線程被創建時，它只會短暫地處于這種狀態。此時它已經分配了必須的系統資源，并執行了初始化。此刻線程已經有資格獲取CPU時間了，之后調度器將把這個線程轉變為可運行狀態或阻塞狀態。
2.就緒(Runnable)：在這種狀態下，只要調度器把時間片分配給線程，線程就可以運行。也就是說，在任意時刻，此狀態的線程可以運行也可以不運行。不同于死亡和阻塞狀態。
3.阻塞(Blocked)：線程能夠運行，但有某個條件阻止它的運行。當線程處于阻塞狀態時，調度器將忽略線程，不會分配給線程任何CPU時間。直到線程重新進入了就緒狀態，它才有可能執行操作。
4.死亡(Dead)：處于死亡或終止狀態的線程將不再是可調度的，并且再也不會得到CPU時間，它的任務已結束，或不再是可運行的。任務死亡的通常方式是從run()方法返回，但是任務的線程還可以被中斷，中斷也是屬于死亡。
進入阻塞狀態
一個任務進入阻塞狀態，可能要有如下原因：
通過調用sleep(milliseconds)使任務進入休眠狀態，在這種情況下，任務在指定的時間內不會運行。
通過調用wait()使線程掛起。直到線程得到了notify()或notifyAll()消息（或者在Java SE5的java.util.concurrent類庫中等價的signal()或signalAll()消息），線程才會進入就緒狀態。
任務在等待某個輸入/輸出完成。
任務試圖在某個對象上調用其同步控制方法，但是對象鎖不可用，因為另一個任務已經獲取了這個鎖。
在較早的代碼中，會有suspend()和resume()來阻塞和喚醒線程，因為容易導致死鎖，所以被廢止了。
中斷
在阻塞狀態的線程，可以通過中斷來終止該阻塞的任務。Thread類包含interrupt()方法來中斷。如果使用Executor，則使用Future的cancel()來中斷任務。其實Executor的shutdownNow()方法，就是將發送一個interrupt()調用給它所啟動的所有線程。
package com.jc.concurrency;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
* 中斷處于阻塞狀態的線程例子
* 發現只有sleep()操作的才能中斷，其余的io和同步都不能被中斷
* @author
*
*/
class SleepBlocked implements Runnable {
public void run() {
try {
TimeUnit.SECONDS.sleep(100);
} catch (InterruptedException e) {
System.out.println("InterruptedException");
}
System.out.println("Exiting SleepBlocked.run()");
}
}
class IOBlocked implements Runnable {
private InputStream in;
public IOBlocked(InputStream is) {
in = is;
}
public void run() {
try {
System.out.println("Waiting for read():");
in.read();
} catch (IOException e) {
if (Thread.currentThread().isInterrupted()) {
System.out.println("Interrupted from blocked I/O");
} else {
throw new RuntimeException(e);
}
}
System.out.println("Exiting IOBlocked.run()");
}
}
class SynchronizedBlocked implements Runnable {
public synchronized void f() {
while (true) // Never releases lock
Thread.yield();
}
public SynchronizedBlocked() {
new Thread() {
public void run() {
f(); // Lock acquired by this thread
}
}.start();
}
public void run() {
System.out.println("Trying to call f()");
f();
System.out.println("Exiting SynchronizedBlocked.run()");
}
}
public class Interrupting {
private static ExecutorService exec = Executors.newCachedThreadPool();
static void test(Runnable r) throws InterruptedException {
Future f = exec.submit(r);
TimeUnit.MILLISECONDS.sleep(100);
System.out.println("Interrupting " + r.getClass().getName());
f.cancel(true); // Interrupts if running
System.out.println("Interrupt sent to " + r.getClass().getName());
}
public static void main(String[] args) throws Exception {
test(new SleepBlocked());
test(new IOBlocked(System.in));
test(new SynchronizedBlocked());
TimeUnit.SECONDS.sleep(3);
System.out.println("Aborting with System.exit(0)");
System.exit(0); // ... since last 2 interrupts failed
}
}
發現只有sleep()操作的才能中斷，其余的io和同步都不能被中斷。所以有個比較不優雅，但有效的關閉方式：
package com.jc.concurrency;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 中斷IO阻塞的線程的方式：關閉資源
* @author
*
*/
public class CloseResource {
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
ServerSocket server = new ServerSocket(8080);
InputStream socketInput = new Socket("localhost", 8080).getInputStream();
exec.execute(new IOBlocked(socketInput));
exec.execute(new IOBlocked(System.in));
TimeUnit.MILLISECONDS.sleep(100);
System.out.println("Shutting down all threads");
exec.shutdownNow();
TimeUnit.SECONDS.sleep(1);
System.out.println("Closing " + socketInput.getClass().getName());
socketInput.close(); // Releases blocked thread
TimeUnit.SECONDS.sleep(1);
System.out.println("Closing " + System.in.getClass().getName());
System.in.close(); // Releases blocked thread
}
}
之所以要sleep，是想要interrupt都傳到各個線程里面。以達到中斷的效果。
NIO提供了優雅的I/O中斷。
/**
* NIO提供了優雅的I/O中斷
* @author
*
*/
class NIOBlocked implements Runnable {
private final SocketChannel sc;
public NIOBlocked(SocketChannel sc) {
this.sc = sc;
}
public void run() {
try {
System.out.println("Waiting for read() in " + this);
sc.read(ByteBuffer.allocate(1));
} catch (ClosedByInterruptException e) {
System.out.println("ClosedByInterruptException" + this);
} catch (AsynchronousCloseException e) {
System.out.println("AsynchronousCloseException" + this);
} catch (IOException e) {
throw new RuntimeException(e);
}
System.out.println("Exiting NIOBlocked.run() " + this);
}
}
public class NIOInterruption {
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
ServerSocket server = new ServerSocket(8080);
InetSocketAddress isa = new InetSocketAddress("localhost", 8080);
SocketChannel sc1 = SocketChannel.open(isa);
SocketChannel sc2 = SocketChannel.open(isa);
System.out.println(sc1);
System.out.println(sc2);
Future f = exec.submit(new NIOBlocked(sc1));
exec.execute(new NIOBlocked(sc2));
exec.shutdown();
TimeUnit.SECONDS.sleep(1);
// Produce an interrupt via cancel:
f.cancel(true);
TimeUnit.SECONDS.sleep(1);
// Release the block by closing the channel:
sc2.close();
}
}
SleepBlocked例子展示了synchronized的鎖是不可以中斷，這是很危險的。所以ReentrantLock提供了可中斷的能力
package com.jc.concurrency;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* SleepBlocked例子展示了synchronized的鎖是不可以中斷，這是很危險的。
* 所以ReentrantLock提供了可中斷的能力
* @author
*
*/
class BlockedMutex {
private Lock lock = new ReentrantLock();
public BlockedMutex() {
// Acquire it right away, to demonstrate interruption
// of a task blocked on a ReentrantLock:
lock.lock();
}
public void f() {
try {
// This will never be available to a second task
lock.lockInterruptibly(); // Special call
System.out.println("lock acquired in f()");
} catch (InterruptedException e) {
System.out.println("Interrupted from lock acquisition in f()");
}
}
}
class Blocked2 implements Runnable {
BlockedMutex blocked = new BlockedMutex();
public void run() {
System.out.println("Waiting for f() in BlockedMutex");
blocked.f();
System.out.println("Broken out of blocked call");
}
}
public class Interrupting2 {
public static void main(String[] args) throws Exception {
Thread t = new Thread(new Blocked2());
t.start();
TimeUnit.SECONDS.sleep(1);
System.out.println("Issuing t.interrupt()");
t.interrupt();
}
}/**output:
Waiting for f() in BlockedMutex
Issuing t.interrupt()
Interrupted from lock acquisition in f()
Broken out of blocked call
**/
在沒有阻塞的語句時，通過Thread.interrupted()判斷線程被中斷:
package com.jc.concurrency;
import java.util.concurrent.TimeUnit;
/**
* 在沒有阻塞的語句時，通過Thread.interrupted()判斷線程被中斷
* @author
*
*/
class NeedsCleanup {
private final int id;
public NeedsCleanup(int ident) {
id = ident;
System.out.println("NeedsCleanup " + id);
}
public void cleanup() {
System.out.println("Cleaning up " + id);
}
}
class Blocked3 implements Runnable {
private volatile double d = 0.0;
public void run() {
// try {
while (!Thread.interrupted()) {
// point1
NeedsCleanup n1 = new NeedsCleanup(1);
// Start try-finally immediately after definition
// of n1, to guarantee proper cleanup of n1:
try {
System.out.println("Sleeping");
// TimeUnit.SECONDS.sleep(1);
// point2
NeedsCleanup n2 = new NeedsCleanup(2);
// Guarantee proper cleanup of n2:
try {
System.out.println("Calculating");
// A time-consuming, non-blocking operation:
for (int i = 1; i < 2500000; i++)
d = d + (Math.PI + Math.E) / d;
System.out.println("Finished time-consuming operation");
} finally {
n2.cleanup();
}
} finally {
n1.cleanup();
}
}
System.out.println("Exiting via while() test");
// } catch (InterruptedException e) {
// System.out.println("Exiting via InterruptedException");
// }
}
}
public class InterruptingIdiom {
public static void main(String[] args) throws Exception {
if (args.length != 1) {
System.out.println("usage: java InterruptingIdiom delay-in-mS");
System.exit(1);
}
Thread t = new Thread(new Blocked3());
t.start();
TimeUnit.MILLISECONDS.sleep(new Integer(args[0]));
t.interrupt();
}
}
線程協作wait()和notify()
wait()、notify()以及nofityAll()有一個比較特殊的方面，那就是這些方法都是基類Object的方法，而不是Thread的一部分。一開始或許有這種困惑，覺得很奇怪。明明是線程的功能，為啥要放在Object里。那時因為這些方法需要操作鎖，當一個任務在方法里遇到wait()的調用時，線程的執行被掛起（阻塞狀態），對象上的鎖會被是否。因此wait()方法需放在同步控制塊里（與之相對比是sleep()因為不用操作鎖，所以可以放在非同步控制塊里，而且還是Thread的方法）。如果在非同步控制調用這些方法，程序能通過編譯，但運行時會拋IllegalMonitorStateException差異。例子：
package com.jc.concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* wait()和notifyAll()例子，notifyAll會將該對象的wait()方法所阻塞的線程
* @author
*
*/
class Car {
private boolean waxOn = false;
public synchronized void waxed() {
waxOn = true; // Ready to buff
notifyAll();
}
public synchronized void buffed() {
waxOn = false; // Ready for another coat of wax
notifyAll();
}
public synchronized void waitForWaxing() throws InterruptedException {
while (waxOn == false)
wait();
}
public synchronized void waitForBuffing() throws InterruptedException {
while (waxOn == true)
wait();
}
}
class WaxOn implements Runnable {
private Car car;
public WaxOn(Car c) {
car = c;
}
public void run() {
try {
while (!Thread.interrupted()) {
System.out.print("Wax On! ");
TimeUnit.MILLISECONDS.sleep(200);
car.waxed();
car.waitForBuffing();
}
} catch (InterruptedException e) {
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax On task");
}
}
class WaxOff implements Runnable {
private Car car;
public WaxOff(Car c) {
car = c;
}
public void run() {
try {
while (!Thread.interrupted()) {
car.waitForWaxing();
System.out.print("Wax Off! ");
TimeUnit.MILLISECONDS.sleep(200);
car.buffed();
}
} catch (InterruptedException e) {
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax Off task");
}
}
public class WaxOMatic {
public static void main(String[] args) throws Exception {
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOff(car));
exec.execute(new WaxOn(car));
TimeUnit.SECONDS.sleep(5); // Run for a while...
exec.shutdownNow(); // Interrupt all tasks
}
}
notify()和nofityAll()
因為可能有多個任務在單個Car對象上處于wait()狀態，因此調用nofityAll()比只調用notify()要更安全。所以上面那個程序，只有一個任務，因此可以使用notify()來代替notifyAll()。
使用 notify()而不是notifyAll()是一種優化。除非知道notify()會喚醒具體哪個任務，不如還是notifyAll()保守點
在有關Java的線程機制的討論中，有一個令人困惑的描述:notifyAll()將喚醒“所有正在等待的任務”。其實更準確是：當notifyAll()因某個特定鎖而被調用時，只有等待這個鎖的任務才會被喚醒:
package com.jc.concurrency;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 當notifyAll()因某個特定鎖而被調用時，只有等待這個鎖的任務才會被喚醒
* @author
*
*/
class Blocker {
synchronized void waitingCall() {
try {
while (!Thread.interrupted()) {
wait();
System.out.print(Thread.currentThread() + " ");
}
} catch (InterruptedException e) {
// OK to exit this way
}
}
synchronized void prod() {
notify();
}
synchronized void prodAll() {
notifyAll();
}
}
class Task implements Runnable {
static Blocker blocker = new Blocker();
public void run() {
blocker.waitingCall();
}
}
class Task2 implements Runnable {
// A separate Blocker object:
static Blocker blocker = new Blocker();
public void run() {
blocker.waitingCall();
}
}
public class NotifyVsNotifyAll {
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++)
exec.execute(new Task());
exec.execute(new Task2());
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
boolean prod = true;
public void run() {
if (prod) {
System.out.print("
notify() ");
Task.blocker.prod();
prod = false;
} else {
System.out.print("
notifyAll() ");
Task.blocker.prodAll();
prod = true;
}
}
}, 400, 400); // Run every .4 second
TimeUnit.SECONDS.sleep(5); // Run for a while...
timer.cancel();
System.out.println("
Timer canceled");
TimeUnit.MILLISECONDS.sleep(500);
System.out.print("Task2.blocker.prodAll() ");
Task2.blocker.prodAll();
TimeUnit.MILLISECONDS.sleep(500);
System.out.println("
Shutting down");
exec.shutdownNow(); // Interrupt all tasks
}
}
使用wait()和notifyAll()實現生產者和消費者：一個飯店，有一個廚師和一個服務員，這個服務員必須等待廚師準備好食物，當廚師準備好后就會通知服務員，之后服務員上菜，然后服務員繼續等待。
package com.jc.concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 簡單的生產者消費者例子
* 此例子有點局限因為不能有多線程的生產者、多線程的消費者。
* 這例子僅僅展示如果使用wait()和notify()保證有序
* @author
*
*/
class Meal {
private final int orderNum;
public Meal(int orderNum) {
this.orderNum = orderNum;
}
public String toString() {
return "Meal " + orderNum;
}
}
class WaitPerson implements Runnable {
private Restaurant restaurant;
public WaitPerson(Restaurant r) {
restaurant = r;
}
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (this) {
while (restaurant.meal == null)
wait(); // ... for the chef to produce a meal
}
System.out.println("Waitperson got " + restaurant.meal);
synchronized (restaurant.chef) {
restaurant.meal = null;
restaurant.chef.notifyAll(); // Ready for another
}
}
} catch (InterruptedException e) {
System.out.println("WaitPerson interrupted");
}
}
}
class Chef implements Runnable {
private Restaurant restaurant;
private int count = 0;
public Chef(Restaurant r) {
restaurant = r;
}
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (this) {
while (restaurant.meal != null)
wait(); // ... for the meal to be taken
}
if (++count == 10) {
System.out.println("Out of food, closing");
restaurant.exec.shutdownNow();
}
System.out.println("Order up! ");
synchronized (restaurant.waitPerson) {
restaurant.meal = new Meal(count);
restaurant.waitPerson.notifyAll();
}
TimeUnit.MILLISECONDS.sleep(100);
}
} catch (InterruptedException e) {
System.out.println("Chef interrupted");
}
}
}
public class Restaurant {
Meal meal;
ExecutorService exec = Executors.newCachedThreadPool();
WaitPerson waitPerson = new WaitPerson(this);
Chef chef = new Chef(this);
public Restaurant() {
exec.execute(chef);
exec.execute(waitPerson);
}
public static void main(String[] args) {
new Restaurant();
}
}
使用顯式鎖Lock和Condition對象：
package com.jc.concurrency.waxomatic2;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 使用顯式的Lock和Condition對象來修改WaxOMatic例子
* @author
*
*/
class Car {
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
private boolean waxOn = false;
public void waxed() {
lock.lock();
try {
waxOn = true; // Ready to buff
condition.signalAll();
} finally {
lock.unlock();
}
}
public void buffed() {
lock.lock();
try {
waxOn = false; // Ready for another coat of wax
condition.signalAll();
} finally {
lock.unlock();
}
}
public void waitForWaxing() throws InterruptedException {
lock.lock();
try {
while (waxOn == false)
condition.await();
} finally {
lock.unlock();
}
}
public void waitForBuffing() throws InterruptedException {
lock.lock();
try {
while (waxOn == true)
condition.await();
} finally {
lock.unlock();
}
}
}
class WaxOn implements Runnable {
private Car car;
public WaxOn(Car c) {
car = c;
}
public void run() {
try {
while (!Thread.interrupted()) {
System.out.print("Wax On! ");
TimeUnit.MILLISECONDS.sleep(200);
car.waxed();
car.waitForBuffing();
}
} catch (InterruptedException e) {
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax On task");
}
}
class WaxOff implements Runnable {
private Car car;
public WaxOff(Car c) {
car = c;
}
public void run() {
try {
while (!Thread.interrupted()) {
car.waitForWaxing();
System.out.print("Wax Off! ");
TimeUnit.MILLISECONDS.sleep(200);
car.buffed();
}
} catch (InterruptedException e) {
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax Off task");
}
}
public class WaxOMatic2 {
public static void main(String[] args) throws Exception {
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOff(car));
exec.execute(new WaxOn(car));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
}
基于Lock和鏈表存儲結構寫的一個消息隊列：
package com.jc.framework.queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class JcBlockingQueue {
private JcQueueData head;
private JcQueueData tail;
private int size = 0;
private int maxSize = Integer.MAX_VALUE;
private final Lock lock;
private final Condition full;
private final Condition empty;
public JcBlockingQueue() {
lock = new ReentrantLock();
full = lock.newCondition(); //角度是生產者
empty = lock.newCondition(); //角度是消費者
}
public void enQueue(T t) throws InterruptedException {
lock.lock();
if (size == maxSize) {
full.await();
}
if (head == null) {
head = new JcQueueData<>(t, null);
tail = head;
size++;
empty.signalAll();
lock.unlock();
return;
}
JcQueueData jcQueueData = new JcQueueData<>(t, null);
tail.setNext(jcQueueData);
tail = jcQueueData;
size++;
if (size == 1)
empty.signalAll();
lock.unlock();
}
public T deQueue() throws InterruptedException {
lock.lock();
while (head == null) {
empty.await();
}
T t = head.getData();
if (head.next != null) {
JcQueueData next = head.next;
head.next = null;
head = next;
} else {
head = null;
tail = null;
}
size--;
if(size==maxSize-1)
full.signalAll();
lock.unlock();
return t;
}
public int size() {
return size;
}
private class JcQueueData {
private T data;
private JcQueueData next;
public JcQueueData(T data, JcQueueData next) {
this.data = data;
this.next = next;
}
public T getData() {
return data;
}
public void setData(T data) {
this.data = data;
}
public JcQueueData getNext() {
return next;
}
public void setNext(JcQueueData next) {
this.next = next;
}
}
}
ExecutorService的shutdown
ExecutorService的shutdown方法，這有可能還有工作正在執行或準備執行，這情況下，它只是通知線程池再沒有更多任務需要增加到它的內部隊列，而且一旦完成所有等待的工作，就應當關閉。
對應的還有shutdownNow()，此方法中止當前處理中的任務，并忽略所有等待的任務。
參考：《Java編程思想》
文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。
轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/69285.html
摘要：流程源處理源代碼，例如過濾任何值。工藝類從編譯后處理生成的文件，例如對類進行字節碼增強。整合后的測試執行集成測試后執行所需的操作。校驗運行任何檢查以驗證包裝是否有效并符合質量標準。 nodejs和es6 nodejs的語法和es6不一樣，如模塊系統，一個是CommonJS的require、一個是es6的import，寫模塊也不一樣。 nodejs的npm 我來理解，nodejs類似與j...
閱讀 1378·2021-10-08 10:04
閱讀 2681·2021-09-22 15:23
閱讀 2724·2021-09-04 16:40
閱讀 1172·2019-08-29 17:29
閱讀 1492·2019-08-29 17:28
閱讀 2988·2019-08-29 14:02
閱讀 2221·2019-08-29 13:18
閱讀 838·2019-08-23 18:35
