資訊專欄INFORMATION COLUMN
摘要：大多數(shù)保證線程安全的方法是添加各種類型鎖，使用各種同步機(jī)制，用限制對(duì)共享的可變的類變量并發(fā)訪問(wèn)的方式來(lái)保證線程安全。只有保證這兩條語(yǔ)句及中間語(yǔ)句以原子方式執(zhí)行，才能避免多線程覆蓋問(wèn)題。
前言
對(duì)于線程安全，我們有說(shuō)不盡的話題。大多數(shù)保證線程安全的方法是添加各種類型鎖，使用各種同步機(jī)制，用限制對(duì)共享的、可變的類變量并發(fā)訪問(wèn)的方式來(lái)保證線程安全。文本從另一個(gè)角度，使用“比較交換算法”（CompareAndSwap）實(shí)現(xiàn)同樣的需求。我們實(shí)現(xiàn)一個(gè)簡(jiǎn)單的“?！?，并逐步重構(gòu)代碼來(lái)進(jìn)行講解。
本文通俗易懂，不會(huì)涉及到過(guò)多的底層知識(shí)，適合初學(xué)者閱讀（言外之意是各位大神可以繞道了）。
“?！保╯tack）是大家經(jīng)常使用的抽象數(shù)據(jù)類型（啥？！不知道，請(qǐng)自行百度）。“?！睗M足“后進(jìn)先出”特性。我們用鏈表數(shù)據(jù)結(jié)構(gòu)完成一個(gè)簡(jiǎn)單的實(shí)現(xiàn)：
public class Stack{ //鏈表結(jié)構(gòu)頭部節(jié)點(diǎn) private Node head; /** * 入棧 * @param item */ public void push(E item) { //為新插入item創(chuàng)建一個(gè)新node Node newHead = new Node<>(item); if(head!=null){ //將新節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)指向原來(lái)的頭部 newHead.next = head; } //將頭部指向新的節(jié)點(diǎn) head=newHead; } /** * 出棧 * @return */ public E pop() { if(head==null){ //當(dāng)前鏈表為空 return null; } //暫存當(dāng)前節(jié)點(diǎn)。 Node oldHead=head; //將當(dāng)前節(jié)點(diǎn)指向當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn) head=head.next; //從暫存的當(dāng)前節(jié)點(diǎn)記錄返回?cái)?shù)據(jù) return oldHead.item; } /** * 鏈表中的節(jié)點(diǎn) * @param */ private static class Node { //節(jié)點(diǎn)保存的數(shù)據(jù) public final E item; //指向下一個(gè)鏈表中下一個(gè)節(jié)點(diǎn) public Node next; public Node(E item) { this.item = item; } } }
代碼使用鏈表數(shù)據(jù)結(jié)構(gòu)實(shí)現(xiàn)“棧”，在Stack中維護(hù)一個(gè)鏈表的“頭部節(jié)點(diǎn)”，通過(guò)對(duì)頭部節(jié)點(diǎn)的操作完成入棧和出棧操作。
我們運(yùn)行代碼測(cè)試一下：
public static void main(String[] args) {
Stack stack=new Stack<>();
for (int i = 0; i < 3; i++) {
//入棧1、2、3
stack.push(i+1);
}
for (int i = 0; i < 3; i++) {
//出棧3、2、1
System.out.println(stack.pop());
}
}
結(jié)果為：
3 2 1
我們使用入棧方法向Stack插入1、2、3，使用出棧方法打印為3、2、1，符合預(yù)期。
2.讓多線程搗搗亂前面我們已經(jīng)測(cè)試過(guò)我們的方法，符合我們對(duì)Stack功能的預(yù)期，那是不是任何情況先我們的“?！倍寄苷９ぷ髂?？
我們運(yùn)行如下代碼：
public static void main(String[] args) {
Stack stack=new Stack<>();
int max=3;
Thread[] threads=new Thread[max];
for (int i = 0; i < max; i++) {
int temp=i;
//入棧1、2、3
Thread thread=new Thread(new Runnable() {
@Override
public void run() {
stack.push(temp+1);
}
});
thread.start();
threads[temp]=thread;
}
//等待所有線程完成。
for (int i = 0; i < max; i++) {
try {
threads[i].join();
} catch (InterruptedException e) {
}
}
for (int i = 0; i < max; i++) {
//出棧3、2、1
System.out.println(stack.pop());
}
}
你可能運(yùn)行了很多次，每次運(yùn)行時(shí)除了打印順序（3、2、1或2、3、1或1、2、3）有變化之外也沒(méi)有發(fā)現(xiàn)其他異常，你可能會(huì)說(shuō)打印順序變化很正常呀，因?yàn)槲覀兊膶⑷霔２僮鞣诺疆惒骄€程中操作，三個(gè)線程的執(zhí)行過(guò)程由系統(tǒng)調(diào)度，所以入棧操作的內(nèi)容自然每次都有可能不同。
好吧，你說(shuō)的沒(méi)錯(cuò)，至少?gòu)拇罅窟\(yùn)行的結(jié)果上看是這樣的，但是這就是多線程編程的奇（tao）幻(yan)之處，也許你運(yùn)行一次沒(méi)有問(wèn)題，兩次沒(méi)有問(wèn)題，一萬(wàn)次也沒(méi)有問(wèn)題，但是終有一次你會(huì)得到那個(gè)意想不到的結(jié)果（你也不想得到，因?yàn)槟鞘莃ug）。這就像一個(gè)“黑天鵝事件”，小概率但是一定會(huì)發(fā)生，且發(fā)生后對(duì)你的系統(tǒng)影響不堪設(shè)想。
下面讓我?guī)憧纯慈绾蔚玫揭饬现獾慕Y(jié)果：
我們使用調(diào)試模式運(yùn)行上面的程序在Stack中push()方法第一行打一個(gè)斷點(diǎn)，然后按照表格中的順序切換不同的線程以單步調(diào)試（step over）方式運(yùn)行run方法中的每一步，直到遇到Resume。
| 執(zhí)行順序 | thread-0 | thread-1 | thread-2 |
|---|---|---|---|
| 1 | Node |
-- | -- |
| 2 | head=newHead; | -- | -- |
| 3 | (Resume) | -- | -- |
| 4 | -- | Node |
-- |
| 5 | -- | -- | Node |
| 6 | -- | newHead.next = head; | -- |
| 7 | -- | -- | newHead.next = head; |
| 8 | -- | head=newHead; | -- |
| 9 | -- | -- | head=newHead; |
| 10 | -- | (Resume) | |
| 11 | -- | -- | (Resume) |
當(dāng)你再次看到打印結(jié)果，你會(huì)發(fā)現(xiàn)結(jié)果為3、1、null，“黑天鵝”出現(xiàn)了。
異常結(jié)果是如何產(chǎn)生的？
當(dāng)thread-0執(zhí)行到順序3時(shí)，head表示的鏈表為node(1)。
當(dāng)thread-1執(zhí)行到順序10時(shí)，head表示的鏈表為node(2)->node(1)。
當(dāng)thread-2執(zhí)行到順序11時(shí)，head表示的鏈表為node(3)->node(1)。
當(dāng)三個(gè)線程都執(zhí)行完畢之后，head的最終表示為node(3)->node(1)，也就是說(shuō)thread-2將thread-1的執(zhí)行結(jié)果覆蓋了。
語(yǔ)句newHead.next = head;是對(duì)頭部節(jié)點(diǎn)的讀取。語(yǔ)句head=newHead;是對(duì)頭部節(jié)點(diǎn)的寫入操作。這兩條語(yǔ)句組成了一個(gè)“讀取——設(shè)置——寫入”語(yǔ)句模式（就像n=n+1）。
如果一個(gè)線程執(zhí)行了共享頭部變量讀取語(yǔ)句，切換其他線程執(zhí)行了修改共享變量的值，再切回到第一個(gè)線程后，第一個(gè)線程中修改頭部結(jié)點(diǎn)的數(shù)據(jù)就不是最新的數(shù)據(jù)為依據(jù)的，所以修改之后其他線程的修改就被覆蓋了。
只有保證這兩條語(yǔ)句及中間語(yǔ)句以原子方式執(zhí)行，才能避免多線程覆蓋問(wèn)題。
大家可以任意調(diào)整代碼中讀取頭部節(jié)點(diǎn)和寫入頭部節(jié)點(diǎn)的調(diào)試順序，制造多線程交錯(cuò)讀寫觀察不同的異常結(jié)果。
為什么我們直接執(zhí)行無(wú)法看到異常結(jié)果呢？
因?yàn)槲覀兊膔un方法很簡(jiǎn)單，在CPU分配的時(shí)間片內(nèi)能運(yùn)行完，沒(méi)有出現(xiàn)在不同的運(yùn)行周期中交錯(cuò)運(yùn)行的狀態(tài)。所以我們才要用調(diào)試模式這種交錯(cuò)運(yùn)行。
為什么上文中我說(shuō)過(guò)這種異常一定會(huì)發(fā)生？
原因在于我們?cè)赟tack類中對(duì)共享的、可變的變量head進(jìn)行的多線程讀寫操作。
怎么才能保證類Stack在多線程情況下運(yùn)行正確？
引用一段《JAVA并發(fā)編程實(shí)踐》中的話：
無(wú)論何時(shí)，只要有多于一個(gè)的線程訪問(wèn)給定的狀態(tài)變量，而且其中某個(gè)線程會(huì)寫入該變量，此時(shí)必須使用同步來(lái)協(xié)調(diào)線程對(duì)該變量的訪問(wèn)。
好吧，看來(lái)我們必須采用“同步”方法了，來(lái)保障我們的Stack類在多線程并行和單線程串行的情況下都有正確的結(jié)果，也就是說(shuō)將Stack變成一個(gè)線程安全的類。
3.讓你搗亂，請(qǐng)家長(zhǎng)！既然多線程總來(lái)?yè)v亂，我們就請(qǐng)他的家長(zhǎng)，讓家長(zhǎng)管管他，守守規(guī)矩，不在搗亂。
我們已經(jīng)知道了Stack類問(wèn)什么不能再多線程下正確的運(yùn)行的原因，所有我們要限制多線程對(duì)Stack類中head變量的并發(fā)寫入，Stack方法中push()和pop()方法都會(huì)對(duì)head進(jìn)行寫操作，所以要限制這兩個(gè)方法不能多線程并發(fā)訪問(wèn)，所以我們想到了synchronized關(guān)鍵字。
程序重構(gòu)：
public class SynchronizedStack{ //鏈表結(jié)構(gòu)頭部節(jié)點(diǎn) private Node head; /** * 入棧 * @param item */ public synchronized void push(E item) { //為新插入item創(chuàng)建一個(gè)新node Node newHead = new Node<>(item); if(head!=null){ //將新節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)指向原來(lái)的頭部 newHead.next = head; } //將頭部指向新的節(jié)點(diǎn) head=newHead; } /** * 出棧 * @return */ public synchronized E pop() { if(head==null){ //當(dāng)前鏈表為空 return null; } //暫存當(dāng)前節(jié)點(diǎn)。 Node oldHead=head; //將當(dāng)前節(jié)點(diǎn)指向當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn) head=head.next; //從暫存的當(dāng)前節(jié)點(diǎn)記錄返回?cái)?shù)據(jù) return oldHead.item; } /** * 鏈表中的節(jié)點(diǎn) * @param */ private static class Node { //節(jié)點(diǎn)保存的數(shù)據(jù) public final E item; //指向下一個(gè)鏈表中下一個(gè)節(jié)點(diǎn) public Node next; public Node(E item) { this.item = item; } } }
將Stack類替換為SynchronizedStack類的測(cè)試方法。
public static void main(String[] args) {
SynchronizedStack stack=new SynchronizedStack<>();
int max=3;
Thread[] threads=new Thread[max];
for (int i = 0; i < max; i++) {
int temp=i;
//入棧1、2、3
Thread thread=new Thread(new Runnable() {
@Override
public void run() {
stack.push(temp+1);
}
});
thread.start();
threads[temp]=thread;
}
//等待所有線程完成。
for (int i = 0; i < max; i++) {
try {
threads[i].join();
} catch (InterruptedException e) {
}
}
for (int i = 0; i < max; i++) {
//出棧3、2、1
System.out.println(stack.pop());
}
}
我們?cè)俅芜\(yùn)行第二章為多線程準(zhǔn)備的測(cè)試方法，發(fā)現(xiàn)當(dāng)執(zhí)行一個(gè)線程的方法時(shí)，其他線程的方法均被阻塞，只能等到第一個(gè)線程方法執(zhí)行完成之后才能執(zhí)行其他線程方法。
我們只不過(guò)是在push()和pop()方法上加入了synchronized 關(guān)鍵字，就將這兩個(gè)方法編程了同步方法，在多線程并發(fā)的情況下也如同單線程串行調(diào)用一般，方法再不能在線程間交替運(yùn)行，也就不能對(duì)head變量做并發(fā)更改了，這樣修改的Stack類就是線程安全的了。
除了synchronized關(guān)鍵字，還有其他的方式實(shí)現(xiàn)加鎖嗎？
除了synchronized關(guān)鍵字還可以使用java.util.concurrent.locks包中各種鎖來(lái)保證同步，但是大概思路都是相同的，都是使用阻塞其他線程的方式在達(dá)到防止并發(fā)寫入的目的。
阻塞線程是否會(huì)影響執(zhí)行效率？
如果和不加通過(guò)的“?！鳖愊啾?，在多線程執(zhí)行的之后效率一定會(huì)有影響，因?yàn)橥椒椒ㄏ拗屏司€程之間的并發(fā)性，但是為了保證“棧”類的在多線程環(huán)境時(shí)功能正確，我們不得不做出效率和正確性的權(quán)衡。
必須要對(duì)整個(gè)方法加上鎖嗎？
我們上面已經(jīng)分析了需要加鎖的范圍，只要保證讀取頭部節(jié)點(diǎn)和寫入頭部節(jié)點(diǎn)之間的語(yǔ)句原子性就可以。所以我們可以這樣執(zhí)行。
/**
* 入棧
*
* @param item
*/
public void push(E item) {
//為新插入item創(chuàng)建一個(gè)新node
Node newHead = new Node<>(item);
synchronized (this) {
if (head != null) {
//將新節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)指向原來(lái)的頭部
newHead.next = head;
}
//將頭部指向新的節(jié)點(diǎn)
head = newHead;
}
}
/**
* 出棧
*
* @return
*/
public E pop() {
synchronized (this) {
if (head == null) {
//當(dāng)前鏈表為空
return null;
}
//暫存當(dāng)前節(jié)點(diǎn)。
Node oldHead = head;
//將當(dāng)前節(jié)點(diǎn)指向當(dāng)前節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)
head = head.next;
//從暫存的當(dāng)前節(jié)點(diǎn)記錄返回?cái)?shù)據(jù)
return oldHead.item;
}
}
通過(guò)synchronized 塊實(shí)現(xiàn)，因?yàn)榉椒ū容^簡(jiǎn)單，所以也沒(méi)有很明顯的縮小加鎖范圍。
除了加鎖的方式，是否還有其他方式？
當(dāng)然，我們還有無(wú)鎖化編程來(lái)解決線程之間同步的問(wèn)題。這就是下面要介紹的比較交換算法。
加鎖實(shí)現(xiàn)線程同步的方式是預(yù)防性方式。無(wú)論共享變量是否會(huì)被并發(fā)修改，我們都只允許同一時(shí)刻只有一個(gè)線程運(yùn)行方法來(lái)阻止并發(fā)發(fā)生。這就相當(dāng)于我們假設(shè)并發(fā)一定會(huì)發(fā)生，所以比較悲觀。
現(xiàn)在我們換一種思路，樂(lè)觀一點(diǎn)，不要假設(shè)對(duì)變量的并發(fā)修改一定發(fā)生，這樣也就不用對(duì)方法加鎖阻止多線程并行運(yùn)行方法了。但是一旦發(fā)生了并發(fā)修改，我們想法發(fā)解決就是了，解決的方法就是將這個(gè)操作重試一下。
繼續(xù)重構(gòu)“?！贝a：
