併發基礎
前言
跟着 The Java Tutorials 把併發的一些基礎過了一遍，發現仍然仍是有不少不清楚的地方，主要是由於日常沒有機會實際應用吧，理論知識要有，實踐也很重要，哪怕是寫些小 demo 也能夠的。html
雖然主要是跟着 tutorials 的 concurrency 章節整理的，但這並非官方文檔的一個翻譯哈，看到一個地方有前置技能不足的時候，就會穿插着一些對 API 的學習之類的。java
還有就是。。segmentfault 對 markdown 的解析和我用的編輯器好像不太一致，引用的地方有點亂，實在懶得改了。。要是看着不舒服，能夠看這個 併發基礎（一）git
Concurrency Foundation
1. Synchronized
Java 提供了兩種基礎的同步語法：1. synchronized 方法 2. synchronized 語句塊github
同步是在內置鎖（intrinsic lock）或者叫作監視鎖（monitor lock）的基礎上創建的（API 中一般將其稱爲 monitor），這個 monitor 在同步的兩個方面發揮做用： 1. 強制性的單獨訪問對象 2. 創建 happens-before 關係編程
每一個對象都有本身的內置鎖，若是有線程想要訪問這個對象，須要先獲取這個對象的內置鎖，那麼此時其餘線程是沒法獲取這個鎖的，也就是沒法訪問這個對象，直至先前的線程釋放鎖。segmentfault
synchronized 就是 Java 對內置鎖的支持。api
- synchronized 方法
當線程調用 synchronized 方法時，就會自動獲取該方法對象的 synchronized 鎖，返回時纔會釋放，即便是由沒被 catch 的異常返回的，也會釋放鎖。數組
那麼，對於 static synchronized method 呢？這個讓人有些迷惑，由於 static method 是和類關聯的，而不是對象。緩存
Intrinsic Locks and Synchronization (The Java™ Tutorials > Essential Classes > Concurrency)安全
In this case, the thread acquires the intrinsic lock for the Class object associated with the class. Thus access to class's static fields is controlled by a lock that's distinct from the lock for any instance of the class.
根據文檔的描述，這種狀況下，對類的靜態域的訪問和對類的普通實例的訪問所獲取的鎖是不一樣的。
其實就是類鎖和對象鎖的區別，類鎖也就是 static synchronized method 和 synchronized(xx.class){} 所使用的鎖，對象鎖就是 non-static synchronized method 和 synchronized(xxx){} 所使用的鎖。
由於這兩個鎖是不一樣的，因此若是對同一個類 A，線程 1 想獲取類 A 對象實例的鎖，和線程 2 想獲取類 A 的類鎖，這兩個線程是不存在競爭關係的。
下面看看 synchronized method 的具體做用：
-
做用：
- 同一個對象上的兩個 synchronized 方法的調用不會交替進行，只有一個線程釋放後其餘線程才能獲取這個鎖
- 當一個 synchronized 方法退出後，它會自動和後面再對這個對象的 synchrnoized 方法的調用創建一個 happens-before 關係，這能夠保證這個對象的狀態的改變對其餘線程均可見。
- 須要注意的是，構造器是不能被同步的，這是語法上面的要求，這好理解，構造器原本就是隻有建立對象的線程才能訪問，因此不須要同步。
這個 warning 值得注意
Synchronized Methods (The Java™ Tutorials > Essential Classes > Concurrency)
Warning: When constructing an object that will be shared between threads, be very careful that a reference to the object does not "leak" prematurely. For example, suppose you want to maintain aListcalledinstancescontaining every instance of class. You might be tempted to add the following line to your constructor:
<div class="codeblock"><pre>
instances.add(this);
</pre></div>But then other threads can use
instancesto access the object before construction of the object is complete.
這是在多線程中很容易忽略的一個問題，就是在尚未構建對象完成時，其餘線程就已經訪問了這個對象。
- synchronized 算是一種簡單直接的方式去解決多線程之間的干擾和內存一致性錯誤，可是可能會帶來活躍性的問題。
- synchronized 語句和可重入的同步
這部分以及活躍性問題在以前的博客裏有詳細寫了 線程安全性
2. Atomic Access
原子性的動做是一次完成的
Atomic Access (The Java™ Tutorials > Essential Classes > Concurrency)
<ul><li>Reads and writes are atomic for reference variables and for most primitive variables (all types exceptlonganddouble).</li><li>Reads and writes are atomic for all variables declaredvolatile(includinglonganddoublevariables).</li></ul>
翻譯過來就是：
- 對引用變量和除了 long 和 double 的基本數據類型的讀、寫都是原子操做
- 對 volatile 修飾的變量的讀、寫也是原子操做。
這段我看了以後有點迷，由於以前的概念中就是 volatile 只能保證被修飾變量的可見性而不能保證原子性，爲什麼文檔中說 "Reads and writes are atomic for all variables declared volatile"？還有就是，對引用變量的讀寫是原子操做？
想了想，我以爲他想表達的意思是這樣的：
首先，文檔中說的讀寫，確定指的是單獨的讀、單獨的寫，好比 int a = 1; 這句確定是一個原子操做，就是向 a 寫入值 1，這很容易看出來。
可若是是 b = a; 呢？這樣就很容易讓人感到迷惑了，猛然一看，這就是對引用變量的寫入，可是，它應該是分爲幾步操做的，1. 讀取 a 2. 寫入 b，若是說單獨對 a 的讀取和單獨對 b 的寫入，這些都是原子操做，文檔中說的是沒有錯的，但是其實是對於 a 這個變量引用，它是隨時有可能被更新的，也就是說 b = a; 可能被分解爲 1. 讀取 a 2. 寫入 b 3. 寫入 a，這個時候 b = a; 這個語句就會帶來錯誤。
由於這種狀況實際上是很容易遇到的，因此給個人印象就是 b = a; 這種對引用變量的讀寫一般並非原子操做，因此看到文檔中這段話感到很迷惑，文檔的描述是正確的，只不過他所界定的原子操做和讀寫的概念，指的是一種最窄的概念，而且，對於 b = a; 這種對引用變量的賦值，之因此出現問題，其實是由於對變量 a 可見性沒有保證，這就不能讓原子操做背這個鍋了。
文檔中後面也說到了，雖然原子操做是完整的操做，使用它們能夠不用擔憂線程干擾，可是有時仍然須要對原子操做進行同步，由於即便是原子操做也避免不了可能出現的內存一致性錯誤。
Atomic actions cannot be interleaved, so they can be used without fear of thread interference. However, this does not eliminate all need to synchronize atomic actions, because memory consistency errors are still possible.
因此說，使用簡單的原子操做訪問會比用 synchronized 訪問變量會更有效，可是這就須要咱們考慮到內存一致性的問題，所以須要使用哪一種方式訪問變量，就要看應用的規模和複雜程度了。
3. Guarded blocks
多線程之間確定少不了協同工做，最多見的方式就是使用 Guarded block：
public void guardedJoy() {
// Simple loop guard. Wastes
// processor time. Don't do this!
while(!joy) {}
System.out.println("Joy has been achieved!");
}
這就是一個 Guarded block，它會不斷地檢查 joy 這個值，而 joy 是由另外一個線程所設置的。
不過，像上面這樣的循環等待對資源也太浪費了，能夠採用另外一種方式：
public synchronized void guardedJoy() {
// This guard only loops once for each special event, which may not
// be the event we're waiting for.
while(!joy) {
try {
wait();
} catch (InterruptedException e) {}
}
System.out.println("Joy and efficiency have been achieved!");
}
這裏使用到了 Object.wait()
Object (Java Platform SE 8 )
Causes the current thread to wait until another thread invokes the notify() method or the notifyAll() method for this object. In other words, this method behaves exactly as if it simply performs the call wait(0).
它的做用就是讓調用該方法的線程進入 WAITING 狀態，直到有其餘線程調用該對象的 notify() or notifyAll() 方法
The current thread must own this object's monitor. The thread releases ownership of this monitor and waits until another thread notifies threads waiting on this object's monitor to wake up either through a call to the notify method or the notifyAll method. The thread then waits until it can re-obtain ownership of the monitor and resumes execution.
調用 wait() 以前須要先獲取這個對象的鎖，一旦調用以後就會釋放對象的鎖，而後等待，直到其餘線程調用 notify() or notifyAll()，但也不是能讓等待的線程立馬從 wait() 返回，而是要等到調用 notify() or notifyAll() 的線程釋放鎖以後，等待線程纔有機會從 wait() 返回，這個機會就是：當前線程能夠從新獲取鎖的持有權限，而後才能繼續執行。
仍是回到 guardedJoy 那個例子，這個版本的 guardedJoy 變成了 synchronized 的了，爲何呢？剛剛介紹 wait() 方法時說到了，一個線程想要調用一個對象的 wait() 方法首先要獲取到這個對象的鎖，而獲取鎖的最簡單的方式就是在 synchronized 方法裏調用 wait() 啦
那麼，當 Thread1 調用 wait() 時，就會釋放鎖而後掛起。當其餘線程 Thread2 請求獲取這個鎖併成功後調用 notifyAll() 通知等待的全部線程，在 Thread2 釋放這個鎖後，等待的線程 Thread1 再次申請就能夠從新得到這個鎖，以後就能夠從 wait() 方法返回繼續執行了。
public synchronized notifyJoy() {
joy = true;
notifyAll();
}
關於 notify()，這個方法只會喚醒一個線程，而且不容許指定喚醒哪一個線程，這是可能會發生死鎖的。以生產者消費者問題爲例，假設分別有 2 個consumer 和 producer，緩衝區大小爲 1，有可能你喚醒的那個線程可能剛好是相同角色的線程，也就是說如今多是一個 consumer 喚醒了另外一個 consumer，原本 consumer 想要喚醒的是 producer，緩存區仍然爲空的，可是 producer 卻還在 wait，由於錯過了被 consumer 喚醒的機會，從而就會產生死鎖。
那麼何時可使用 notify() 呢？文檔中是這樣給的建議：
Guarded Blocks (The Java™ Tutorials >Essential Classes > Concurrency)
Note: There is a second notification method, notify, which wakes up a single thread. Because notify doesn't allow you to specify the thread that is woken up, it is useful only in massively parallel applications — that is, programs with a large number of threads, all doing similar chores. In such an application, you don't care which thread gets woken up.
因此 notify() 通常只在大規模併發應用（即系統有大量類似任務的線程）中使用。由於對於大規模併發應用，咱們其實並不關心哪個線程被喚醒。
看一下官網給的生產者消費者的示例程序：
數據經過 Drop 對象共享信息：
public class Drop {
// Message sent from producer
// to consumer.
private String message;
// true 表示爲空，須要等待生產數據
// false 表示能夠取數據了
private boolean empty = true;
public synchronized String take() {
// Wait until message is
// available.
while (empty) {
try {
wait();
} catch (InterruptedException e) {}
}
// Toggle status.
empty = true;
// Notify producer that
// status has changed.
notifyAll();
return message;
}
public synchronized void put(String message) {
// Wait until message has
// been retrieved.
while (!empty) {
try {
wait();
} catch (InterruptedException e) {}
}
// Toggle status.
empty = false;
// Store message.
this.message = message;
// Notify consumer that status
// has changed.
notifyAll();
}
}
生產者：
import java.util.Random;
public class Producer implements Runnable {
private Drop drop;
public Producer(Drop drop) {
this.drop = drop;
}
public void run() {
String importantInfo[] = {
"Mares eat oats",
"Does eat oats",
"Little lambs eat ivy",
"A kid will eat ivy too"
};
Random random = new Random();
for (int i = 0;
i < importantInfo.length;
i++) {
drop.put(importantInfo[i]);
try {
// 隨機的暫停一段時間，接近實際中狀況
Thread.sleep(random.nextInt(5000));
} catch (InterruptedException e) {}
}
drop.put("DONE");
}
}
消費者：
import java.util.Random;
public class Consumer implements Runnable {
private Drop drop;
public Consumer(Drop drop) {
this.drop = drop;
}
public void run() {
Random random = new Random();
for (String message = drop.take();
! message.equals("DONE");
message = drop.take()) {
System.out.format("MESSAGE RECEIVED: %s%n", message);
try {
Thread.sleep(random.nextInt(5000));
} catch (InterruptedException e) {}
}
}
}
主線程
public class ProducerConsumerExample {
public static void main(String[] args) {
Drop drop = new Drop();
(new Thread(new Producer(drop))).start();
(new Thread(new Consumer(drop))).start();
}
}
4. Immutable Objects
在併發編程中，一種被廣泛承認的原則就是：儘量的使用不可變對象來建立簡單、可靠的代碼。
由於 immutable objects 建立後不能被修改，因此不會出現因爲線程干擾產生的錯誤 or 內存一致性錯誤。
但有時候會有這種擔心，就是使用 immutable objects 每次建立新對象的開銷會不會太大，若是不使用 immutable objects，就能夠避免建立過多新對象，只須要 update 就能夠。
而文檔中說到，這種建立對象的開銷經常被過度高估，由於使用不可變對象所帶來的一些效率提高能夠抵消這種開銷。
e.g. 使用不可變對象下降了垃圾回收所產生的額外開銷，同時也能夠減小一些爲了維護在併發中的 mutable objects 的代碼開銷。
看一下舉得這個例子：
public class SynchronizedRGB {
// Values must be between 0 and 255.
private int red;
private int green;
private int blue;
private String name;
private void check(int red,
int green,
int blue) {
if (red < 0 || red > 255
|| green < 0 || green > 255
|| blue < 0 || blue > 255) {
throw new IllegalArgumentException();
}
}
public SynchronizedRGB(int red,
int green,
int blue,
String name) {
check(red, green, blue);
this.red = red;
this.green = green;
this.blue = blue;
this.name = name;
}
public void set(int red,
int green,
int blue,
String name) {
check(red, green, blue);
synchronized (this) {
this.red = red;
this.green = green;
this.blue = blue;
this.name = name;
}
}
public synchronized int getRGB() {
return ((red << 16) | (green << 8) | blue);
}
public synchronized String getName() {
return name;
}
public synchronized void invert() {
red = 255 - red;
green = 255 - green;
blue = 255 - blue;
name = "Inverse of " + name;
}
}
假如如今 線程1 在執行下列代碼，已經執行到 Statement 1，而後 線程2 剛好也在執行，可是 線程2 恰恰是執行到 Statement 1 和 2 之間，此時 線程1 若是再繼續執行 Statement 2，就會出現 getRGB() 和 getName() 結果不匹配的狀況，這就是在併發中很容易出現的問題。
SynchronizedRGB color =
new SynchronizedRGB(0, 0, 0, "Pitch Black");
...
int myColorInt = color.getRGB(); //Statement 1
String myColorName = color.getName(); //Statement 2
此時加上了 synchronized，將 Statement 1 和 2 綁定到了一塊兒，這樣併發就不會出問題了。
synchronized (color) {
int myColorInt = color.getRGB();
String myColorName = color.getName();
}
會出現上面這種不匹配的狀況，是由於 color 是一個 mutable object，若是它變成 immutable object，就不會出現這種問題了。
對於如何定義 immutable objects，文檔給出了一個 strategy
A Strategy for Defining Immutable Objects (The Java™ Tutorials >Essential Classes > Concurrency)
<ol><li>Don't provide "setter" methods methods that modify fields or objects referred to by fields.</li><li>Make all fieldsfinalandprivate.</li><li>Don't allow subclasses to override methods. The simplest way to do this is to declare the class asfinal. A more sophisticated approach is to make the constructorprivateand construct instances in factory methods.</li><li>If the instance fields include references to mutable objects, don't allow those objects to be changed:<ul><li>Don't provide methods that modify the mutable objects.</li><li>Don't share references to the mutable objects. Never store references to external, mutable objects passed to the constructor; if necessary, create copies, and store references to the copies. Similarly, create copies of your internal mutable objects when necessary to avoid returning the originals in your methods.</li></ul></li></ol>
最後一點，關於對 mutable objects 的引用：若是一個對外部可變對象的引用須要被傳到構造函數中，必定不要保存這個引用，若是必需要保存，就作一個該對象的 copy，而後保存這個 copy。相似的，若是是引用內部的可變對象，必要時也要建立內部可變對象的 copy，以免在方法中返回原對象引用。
這一點是很容易被忽略的，在 Core Java 中也提到了相似的狀況，做者的建議也是作 copy，不要直接引用原對象，由於很難知道這個對象引用在其餘地方是否是會被改變。
根據上面的 strategy，從新定義 RGB：
final public class ImmutableRGB {
// Values must be between 0 and 255.
final private int red;
final private int green;
final private int blue;
final private String name;
private void check(int red,
int green,
int blue) {
if (red < 0 || red > 255
|| green < 0 || green > 255
|| blue < 0 || blue > 255) {
throw new IllegalArgumentException();
}
}
public ImmutableRGB(int red,
int green,
int blue,
String name) {
check(red, green, blue);
this.red = red;
this.green = green;
this.blue = blue;
this.name = name;
}
public int getRGB() {
return ((red << 16) | (green << 8) | blue);
}
public String getName() {
return name;
}
public ImmutableRGB invert() {
return new ImmutableRGB(255 - red,
255 - green,
255 - blue,
"Inverse of " + name);
}
}
High Level Concurrency Objects
1. Lock Objects
在 liveness 那一節中有舉了一個 Alphonse 和 Gaston 鞠躬產生死鎖的例子，這裏產生死鎖的緣由在於可能 線程1 進入 bow，線程2 進入 bowBack，當 線程1 進入 bow 裏的 bowBack 時，線程2 剛好也正在進入 bowBack，兩者都在等彼此退出，可是卻又永遠不會退出，由於彼此在循環等待，會一直阻塞在這裏，從而產生死鎖（循環等待、不可剝奪、獨自佔有、保持請求）。
public class Deadlock {
static class Friend {
private final String name;
public Friend(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
public synchronized void bow(Friend bower) {
System.out.format("%s: %s"
+ " has bowed to me!%n",
this.name, bower.getName());
bower.bowBack(this);
}
public synchronized void bowBack(Friend bower) {
System.out.format("%s: %s"
+ " has bowed back to me!%n",
this.name, bower.getName());
}
}
public static void main(String[] args) {
final Friend alphonse =
new Friend("Alphonse");
final Friend gaston =
new Friend("Gaston");
new Thread(new Runnable() {
public void run() { alphonse.bow(gaston); }
}).start();
new Thread(new Runnable() {
public void run() { gaston.bow(alphonse); }
}).start();
}
}
在這一節做者用 lock objects 來解決這個問題
先簡單瞭解一下 Lock 這個接口：
Lock Objects (The Java™ Tutorials >Essential Classes > Concurrency)
Lock objects work very much like the implicit locks used by synchronized code. As with implicit locks, only one thread can own a Lock object at a time. Lock objects also support a wait/notify mechanism, through their associated Condition objects.
Lock 很像同步代碼使用的內置鎖，在同一時刻只有一個線程能夠得到 Lock 對象。經過關聯 Condition 對象，Lock 對象也支持 wait/notify 機制。
關於 Condition：
Condition (Java Platform SE 8 )
Where a Lock replaces the use of synchronized methods and statements, a Condition replaces the use of the Object monitor methods.
在 Lock 取代了 synchronized 方法和語句的地方，Condition 相應地取代了 Object 監視器方法（wait, notify and notifyAll）的使用。
看一下 tryLock() 這個方法：
Lock (Java Platform SE 8 )
<!-- --><ul class="blockList"><li class="blockList"><pre>boolean tryLock()</pre><div class="block">Acquires the lock only if it is free at the time of invocation.<p>Acquires the lock if it is available and returns immediately
with the valuetrue.
If the lock is not available then this method will return
immediately with the valuefalse.<p>A typical usage idiom for this method would be:<pre>
Lock lock = ...;
if (lock.tryLock()) {
try {
// manipulate protected state
} finally {
lock.unlock();
}
} else {
// perform alternative actions
}</pre>
This usage ensures that the lock is unlocked if it was acquired, and
doesn't try to unlock if the lock was not acquired.</div>
<dl><dt><span class="returnLabel">Returns:</span></dt><dd>trueif the lock was acquired andfalseotherwise</dd></dl></li></ul>
它會嘗試獲取鎖，若是成功，馬上返回 true，若是失敗也會馬上返回 false，API 文檔中舉得那個例子就是一個典型的用法，這種使用 tryLock() 的方式能夠確保在獲取鎖後必定會釋放鎖，由於 Lock 和 synchronized 有一個不同的地方在於 synchronized 會自動釋放，而 Lock 不會，因此必定要保證手動釋放 Lock 鎖，而且這種方式也能夠保證若是沒獲取鎖的狀況不會 unlock()。
下面看怎麼用 Lock 解決鞠躬的死鎖問題：
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.Random;
public class Safelock {
// 讓 Friend 做爲一個靜態內部類，由於不須要引用外部變量
static class Friend {
private final String name;
// 這裏使用的是可重入鎖
private final Lock lock = new ReentrantLock();
public Friend(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
// 即將到來的鞠躬
public boolean impendingBow(Friend bower) {
Boolean myLock = false;
Boolean yourLock = false;
try {
myLock = lock.tryLock();
yourLock = bower.lock.tryLock();
} finally {
// 若是都獲取到了鎖 or 都沒獲取到，就不會釋放鎖
// 若是隻有一個獲取到了，須要釋放
if (! (myLock && yourLock)) {
if (myLock) {
lock.unlock();
}
if (yourLock) {
bower.lock.unlock();
}
}
}
// 只有兩個鎖都獲取到時纔會返回 true
return myLock && yourLock;
}
public void bow(Friend bower) {
// 若是都獲取到了鎖
if (impendingBow(bower)) {
try {
System.out.format("%s: %s has"
+ " bowed to me!%n",
this.name, bower.getName());
bower.bowBack(this);
} finally {
lock.unlock();
bower.lock.unlock();
}
} else {
// 若是隻獲取到了一個鎖，說明其餘線程搶佔了另外一個鎖
// 在這個情景中就是 Alphonse 鞠躬以前發如今 Gaston 正要鞠躬
// 所以本身就不鞠躬了，從而避免了死鎖
System.out.format("%s: %s started"
+ " to bow to me, but saw that"
+ " I was already bowing to"
+ " him.%n",
this.name, bower.getName());
}
}
public void bowBack(Friend bower) {
System.out.format("%s: %s has" +
" bowed back to me!%n",
this.name, bower.getName());
}
}
// 一樣地，也是將 BowLoop 做爲靜態內部類
static class BowLoop implements Runnable {
private Friend bower;
private Friend bowee;
public BowLoop(Friend bower, Friend bowee) {
this.bower = bower;
this.bowee = bowee;
}
public void run() {
Random random = new Random();
for (;;) {
try {
Thread.sleep(random.nextInt(10));
} catch (InterruptedException e) {}
bowee.bow(bower);
}
}
}
public static void main(String[] args) {
final Friend alphonse =
new Friend("Alphonse");
final Friend gaston =
new Friend("Gaston");
new Thread(new BowLoop(alphonse, gaston)).start();
new Thread(new BowLoop(gaston, alphonse)).start();
}
}
2. Executors
文檔教程只是一些比較簡單的介紹，先過一遍了。
2.1 Executor Interfaces
concurrent 包中主要有三個 Executor 接口
- Executor
- ExecutorService
- ScheduledExecutorService
- Executor
運行新任務。
只有一個 execute() 方法，用來建立線程。可是這個方法沒有定義具體的實現方式，因此對於不一樣的 Executor 的實現，有不一樣的建立方式。
Executor (Java Platform SE 8 )
<!-- --><ul class="blockListLast"><li class="blockList"><h4>execute</h4><pre>void execute( Runnable command)</pre><div class="block">Executes the given command at some time in the future. The command may execute in a new thread, in a pooled thread, or in the calling
thread, at the discretion of theExecutorimplementation.</div><dl><dt><span class="paramLabel">Parameters:</span></dt><dd>command- the runnable task</dd><dt><span class="throwsLabel">Throws:</span></dt><dd>RejectedExecutionException- if this task cannot be
accepted for execution</dd><dd>NullPointerException- if command is null</dd></dl></li></ul></li></ul></li></ul></div></div>
execute() 接受一個 Runnable 對象。
- ExecutorService
ExecutorService 除了提供 execute() 方法還提供了 submit() 方法，這個方法不只能夠接受 Runnable 對象還能夠接受 Callable 對象。Callable 對象可使任務返還執行的結果
Callable (Java Platform SE 8 )
<!-- --><ul class="blockListLast"><li class="blockList"><h4>call</h4><pre>
V call() throws
Exception</pre><div class="block">Computes a result, or throws an exception if unable to do so.</div><dl><dt><span class="returnLabel">Returns:</span></dt><dd>computed result</dd><dt><span class="throwsLabel">Throws:</span></dt><dd>
Exception - if unable to compute a result</dd></dl></li></ul></li></ul></li></ul></div></div>
call() 方法會返回計算的結果。
經過 submit() 方法返回的 Future 對象能夠讀取 Callable 任務的執行結果，或是管理 Callable 任務和 Runnable 任務的狀態。
關於 Future 這個接口，它是表示異步計算的結果，提供的方法是用來 1. 檢查計算是否完成 2. 是否等待計算完成 3. 是否查找計算的結果
簡單使用的例子：
interface ArchiveSearcher {
String search (String target);
}
class App {
ExecutorService executor = ...
ArchiveSearcher searcher = ...
void showSearch(final String target) throws InterruptedException {
Future<String> future = executor.submit(new Callable<String>() {
public String call() {
return searcher.search(target);
}
});
displayOtherThings(); // do other things while searching
try {
// 計算完成時才能用 get() 獲取結果，若是沒有計算完成會阻塞着
displayText(future.get()); // use future
} catch (ExecutionException ex) {
cleanup();
return;
}
}
}
submit 裏的 Callable 對象也能夠替換成以下代碼：
FutureTask<String> future =
new FutureTask<String>(new Callable<String>() {
public String call() {
return searcher.search(target);
}});
executor.execute(future);
由於 FutureTask 是實現了 Runnable 接口的 Future 的實現，因此能夠由 Executor 來執行。
public class FutureTask<V> extends Object implements RunnableFuture<V>
public interface RunnableFuture<V> extends Runnable, Future<V>
ExecutorService 也提供了批量運行 Callable 任務的方法。最後，ExecutorService 還提供了一些關閉執行器的方法。若是須要支持即時關閉，執行器所執行的任務須要正確處理中斷。
- ScheduledExecutorService
擴展了 ExecutorService 接口，添加了 schedule 方法。
public interface ScheduledExecutorService extends ExecutorService
經過 schedule 方法可讓命令在給定延遲的時間以後執行或者按期執行。
2.2 Thread Pools
這裏只是對線程池的一個簡單的介紹。
大多數 concurrent 包裏的 executor 的實現都使用了線程池（由 worker 線程組成），worker 線程獨立於它所執行的 Runnable 任務和 Callable 任務，而且常常用來執行多個任務。
使用 worker 線程可使建立線程的開銷最小化。在大規模併發應用中，建立大量的Thread對象會佔用佔用大量系統內存，分配和回收這些對象會產生很大的開銷。
一種最多見的線程池是 fixed thread pool（固定大小的線程池）。這種線程池始終有必定數量的線程在運行，若是一個線程因爲某種緣由終止運行了，線程池會自動建立一個新的線程來代替它。須要執行的任務經過一個內部隊列提交給線程，當沒有更多的工做線程能夠用來執行任務時，隊列保存額外的任務。 使用 fixed thread pool 的一個很重要的好處是能夠 degrade gracefully。
好比一個 Web 服務器，每個 HTTP 請求都是由一個單獨的線程來處理，不可能爲每個 HTTP 請求都建立一個新線程，這樣的話當系統的開銷超出其能力時，會忽然地對全部請求都中止響應。若是限制 Web 服務器能夠建立的線程數量，那麼它就沒必要當即處理全部收到的請求，而是在有能力處理請求時才處理。
建立一個使用 fixed thread pool 的 executor 的最簡單的方法是調用 java.util.concurrent.Executors 的 newFixedThreadPool 工廠方法。
Executors 還提供了下面的工廠方法：
- newCachedThreadPool
- newSingleThreadExecutor
- 還有一些建立 ScheduledExecutorService executor 的方法。
還有其餘的，如 ThreadPoolExecutor or ScheduledThreadPoolExecutor
2.3 Fork/Join
Basic use:
if (my portion of the work is small enough) do the work directly else split my work into two pieces invoke the two pieces and wait for the results
思想有點像分而治之的感受，主要是能夠充分利用多處理器系統的並行處理能力。
文檔中舉了一個圖片模糊處理的例子：
假設你想要模糊一張圖片。原始的 source 圖片由一個整數的數組表示，每一個整數表示一個像素點的顏色數值。與 source 圖片相同，模糊以後的 destination 圖片也由一個整數數組表示。 對圖片的模糊操做是經過對 source 數組中的每個像素點進行處理完成的。
處理的過程：將每一個像素點的色值取出，與周圍像素的色值（紅、黃、藍）放在一塊兒取平均值，獲得的結果被放入 destination 數組。
由於一張圖片會由一個很大的數組來表示，因此處理圖片過程可能會很耗時，可是若是使用 fork/join 框架來完成，就能夠充分利用多處理器系統的並行處理能力，加快處理速度。
public class ForkBlur extends RecursiveAction {
private int[] mSource;
private int mStart;
private int mLength;
private int[] mDestination;
// Processing window size; should be odd.
private int mBlurWidth = 15;
public ForkBlur(int[] src, int start, int length, int[] dst) {
mSource = src;
mStart = start;
mLength = length;
mDestination = dst;
}
protected void computeDirectly() {
int sidePixels = (mBlurWidth - 1) / 2;
for (int index = mStart; index < mStart + mLength; index++) {
// Calculate average.
float rt = 0, gt = 0, bt = 0;
for (int mi = -sidePixels; mi <= sidePixels; mi++) {
int mindex = Math.min(Math.max(mi + index, 0),
mSource.length - 1);
int pixel = mSource[mindex];
rt += (float)((pixel & 0x00ff0000) >> 16)
/ mBlurWidth;
gt += (float)((pixel & 0x0000ff00) >> 8)
/ mBlurWidth;
bt += (float)((pixel & 0x000000ff) >> 0)
/ mBlurWidth;
}
// Reassemble destination pixel.
int dpixel = (0xff000000 ) |
(((int)rt) << 16) |
(((int)gt) << 8) |
(((int)bt) << 0);
mDestination[index] = dpixel;
}
}
...
如今實現抽象方法 compute()，在處理時，能夠直接計算，也能夠將其分開計算，取決於一個閾值，這個閾值能夠簡單地用數組的長度來表明。
protected static int sThreshold = 100000;
protected void compute() {
if (mLength < sThreshold) {
computeDirectly();
return;
}
int split = mLength / 2;
invokeAll(new ForkBlur(mSource, mStart, split, mDestination),
new ForkBlur(mSource, mStart + split, mLength - split,
mDestination));
}
由於 fork/join 的核心就是 ForkJoinPool 類，ForkJoinPool 能夠執行 ForkJoinTask 任務，而 RecursiveAction 繼承了 ForkJoinTask。因此上面這個方法若是是在 RecursiveAction 類中，就能夠在 ForkJoinPool 中設置任務並令其執行。
// source image pixels are in src // destination image pixels are in dst ForkBlur fb = new ForkBlur(src, 0, src.length, dst); // Create the ForkJoinPool that will run the task. ForkJoinPool pool = new ForkJoinPool(); // Run the task. pool.invoke(fb);
參考資料
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 併發基礎
- 2. JAVA併發基礎
- 3. Java基礎-併發
- 4. java併發基礎
- 5. 併發基礎一
- 6. Java併發基礎
- 7. 高併發 - 基礎
- 8. golang併發基礎
- 9. 高併發基礎
- 10. 【併發編程-基礎】(一)併發編程基礎