啃碎併發（六）：Java線程同步與實現

時間 2019-12-07

標籤併發 java 線程同步實現欄目 Java 简体版

原文原文鏈接

0 前言

爲什麼要使用Java線程同步？ Java容許多線程併發控制，當多個線程同時操做一個可共享的資源變量時，將會致使數據不許確，相互之間產生衝突，所以加入同步鎖以免在該線程沒有完成操做以前，被其餘線程的調用，從而保證了該變量的惟一性和準確性。java

但其併發編程的根本，就是使線程間進行正確的通訊。其中兩個比較重要的關鍵點，以下：編程

線程通訊：重點關注線程同步的幾種方式；

正確通訊：重點關注是否有線程安全問題；

Java中提供了不少線程同步操做，好比：synchronized關鍵字、wait/notifyAll、ReentrantLock、Condition、一些併發包下的工具類、Semaphore，ThreadLocal、AbstractQueuedSynchronizer等。本文主要說明一下這幾種同步方式的使用及優劣。安全

1 ReentrantLock可重入鎖

自JDK5開始，新增了Lock接口以及它的一個實現類ReentrantLock。ReentrantLock可重入鎖是J.U.C包內置的一個鎖對象，能夠用來實現同步，基本使用方法以下：bash

public class ReentrantLockTest {

    private ReentrantLock lock = new ReentrantLock();

    public void execute() {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + " do something synchronize");
            try {
                Thread.sleep(5000l);
            } catch (InterruptedException e) {
                System.err.println(Thread.currentThread().getName() + " interrupted");
                Thread.currentThread().interrupt();
            }
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        thread1.start();
        thread2.start();
    }
}
複製代碼

上面例子表示 同一時間段只能有1個線程執行execute方法，輸出以下：多線程

Thread-0 do something synchronize
// 隔了5秒鐘 輸入下面
Thread-1 do something synchronize
複製代碼

可重入鎖中可重入表示的意義在於 對於同一個線程，能夠繼續調用加鎖的方法，而不會被掛起。可重入鎖內部維護一個計數器，對於同一個線程調用lock方法，計數器+1，調用unlock方法，計數器-1。併發

舉個例子再次說明一下可重入的意思：在一個加鎖方法execute中調用另一個加鎖方法anotherLock並不會被掛起，能夠直接調用(調用execute方法時計數器+1，而後內部又調用了anotherLock方法，計數器+1，變成了2)：dom

public void execute() {
    lock.lock();
    try {
        System.out.println(Thread.currentThread().getName() + " do something synchronize");
        try {
            anotherLock();
            Thread.sleep(5000l);
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    } finally {
        lock.unlock();
    }
}

public void anotherLock() {
    lock.lock();
    try {
        System.out.println(Thread.currentThread().getName() + " invoke anotherLock");
    } finally {
        lock.unlock();
    }
}
複製代碼

輸出：ide

Thread-0 do something synchronize
Thread-0 invoke anotherLock
// 隔了5秒鐘 輸入下面
Thread-1 do something synchronize
Thread-1 invoke anotherLock
複製代碼

2 synchronized

synchronized跟ReentrantLock同樣，也支持可重入鎖。可是它是 一個關鍵字，是一種語法級別的同步方式，稱爲內置鎖：工具

public class SynchronizedKeyWordTest {

    public synchronized void execute() {
            System.out.println(Thread.currentThread().getName() + " do something synchronize");
        try {
            anotherLock();
            Thread.sleep(5000l);
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    }

    public synchronized void anotherLock() {
        System.out.println(Thread.currentThread().getName() + " invoke anotherLock");
    }

    public static void main(String[] args) {
        SynchronizedKeyWordTest reentrantLockTest = new SynchronizedKeyWordTest();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                reentrantLockTest.execute();
            }
        });
        thread1.start();
        thread2.start();
    }

}
複製代碼

輸出結果跟ReentrantLock同樣，這個例子說明內置鎖能夠做用在方法上。synchronized關鍵字也能夠修飾靜態方法，此時若是調用該靜態方法，將會鎖住整個類。性能

同步是一種高開銷的操做，所以應該儘可能減小同步的內容。一般沒有必要同步整個方法，使用synchronized代碼塊同步關鍵代碼便可。

synchronized跟ReentrantLock相比，有幾點侷限性：

加鎖的時候不能設置超時。ReentrantLock有提供tryLock方法，能夠設置超時時間，若是超過了這個時間而且沒有獲取到鎖，就會放棄，而synchronized卻沒有這種功能；

ReentrantLock可使用多個Condition，而synchronized卻只能有1個

不能中斷一個試圖得到鎖的線程；

ReentrantLock能夠選擇公平鎖和非公平鎖；

ReentrantLock能夠得到正在等待線程的個數，計數器等；

因此，Lock的操做與synchronized相比，靈活性更高，並且Lock提供多種方式獲取鎖，有Lock、ReadWriteLock接口，以及實現這兩個接口的ReentrantLock類、ReentrantReadWriteLock類。

關於Lock對象和synchronized關鍵字選擇的考量：

最好兩個都不用，使用一種java.util.concurrent包提供的機制，可以幫助用戶處理全部與鎖相關的代碼。

若是synchronized關鍵字能知足用戶的需求，就用synchronized，由於它能簡化代碼。

若是須要更高級的功能，就用ReentrantLock類，此時要注意及時釋放鎖，不然會出現死鎖，一般在finally代碼釋放鎖。

在性能考量上來講，若是競爭資源不激烈，二者的性能是差很少的，而當競爭資源很是激烈時（即有大量線程同時競爭），此時Lock的性能要遠遠優於synchronized。因此說，在具體使用時要根據適當狀況選擇。

3 Condition條件對象

Condition條件對象的意義在於 對於一個已經獲取Lock鎖的線程，若是還須要等待其餘條件才能繼續執行的狀況下，纔會使用Condition條件對象。

Condition能夠替代傳統的線程間通訊，用await()替換wait()，用signal()替換notify()，用signalAll()替換notifyAll()。

爲何方法名不直接叫wait()/notify()/nofityAll()？由於Object的這幾個方法是final的，不可重寫！

public class ConditionTest {

    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        Condition condition = lock.newCondition();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                    System.out.println(Thread.currentThread().getName() + " wait for condition");
                    try {
                        condition.await();
                        System.out.println(Thread.currentThread().getName() + " continue");
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    lock.unlock();
                }
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                    System.out.println(Thread.currentThread().getName() + " sleep 5 secs");
                    try {
                        Thread.sleep(5000l);
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                        Thread.currentThread().interrupt();
                    }
                    condition.signalAll();
                } finally {
                    lock.unlock();
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}
複製代碼

這個例子中thread1執行到condition.await()時，當前線程會被掛起，直到thread2調用了condition.signalAll()方法以後，thread1纔會從新被激活執行。

這裏須要注意的是thread1調用Condition的await方法以後，thread1線程釋放鎖，而後立刻加入到Condition的等待隊列，因爲thread1釋放了鎖，thread2得到鎖並執行，thread2執行signalAll方法以後，Condition中的等待隊列thread1被取出並加入到AQS中，接下來thread2執行完畢以後釋放鎖，因爲thread1已經在AQS的等待隊列中，因此thread1被喚醒，繼續執行。

傳統線程的通訊方式，Condition均可以實現。Condition的強大之處在於它能夠爲多個線程間創建不一樣的Condition。

注意，Condition是被綁定到Lock上的，要建立一個Lock的Condition必須用newCondition()方法。

4 wait&notify/notifyAll方式

Java線程的狀態轉換圖與相關方法，以下：

在圖中，紅框標識的部分方法，能夠認爲已過期，再也不使用。上圖中的方法可以參與到線程同步中的方法，以下：

wait、notify、notifyAll方法：線程中通訊可使用的方法。線程中調用了wait方法，則進入阻塞狀態，只有等另外一個線程調用與wait同一個對象的notify方法。這裏有個特殊的地方，調用wait或者notify，前提是須要獲取鎖，也就是說，須要在同步塊中作以上操做。

wait/notifyAll方式跟ReentrantLock/Condition方式的原理是同樣的。

Java中每一個對象都擁有一個內置鎖，在內置鎖中調用wait，notify方法至關於調用鎖的Condition條件對象的await和signalAll方法。

public class WaitNotifyAllTest {

    public synchronized void doWait() {
        System.out.println(Thread.currentThread().getName() + " run");
        System.out.println(Thread.currentThread().getName() + " wait for condition");
        try {
            this.wait();
            System.out.println(Thread.currentThread().getName() + " continue");
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    }

    public synchronized void doNotify() {
        try {
            System.out.println(Thread.currentThread().getName() + " run");
            System.out.println(Thread.currentThread().getName() + " sleep 5 secs");
            Thread.sleep(5000l);
            this.notifyAll();
        } catch (InterruptedException e) {
            System.err.println(Thread.currentThread().getName() + " interrupted");
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        WaitNotifyAllTest waitNotifyAllTest = new WaitNotifyAllTest();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                waitNotifyAllTest.doWait();
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                waitNotifyAllTest.doNotify();
            }
        });
        thread1.start();
        thread2.start();
    }
}
複製代碼

這裏須要注意的是 調用wait/notifyAll方法的時候必定要得到當前線程的鎖，不然會發生IllegalMonitorStateException異常。

join方法：該方法主要做用是在該線程中的run方法結束後，才往下執行。

package com.thread.simple;
 
public class ThreadJoin {
    public static void main(String[] args) {
        Thread thread= new Thread(new Runnable() {
              @Override
              public void run() {
                   System.err.println("線程"+Thread.currentThread().getId()+" 打印信息");
              }
        });
        thread.start();
 	
        try {
            thread.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.err.println("主線程打印信息");	
    }
}
複製代碼

yield方法：線程自己的調度方法，使用時線程能夠在run方法執行完畢時，調用該方法，告知線程已能夠出讓CPU資源。

public class Test1 {  
    public static void main(String[] args) throws InterruptedException {  
        new MyThread("低級", 1).start();  
        new MyThread("中級", 5).start();  
        new MyThread("高級", 10).start();  
    }  
}  

class MyThread extends Thread {  
    public MyThread(String name, int pro) {  
        super(name);// 設置線程的名稱 
        this.setPriority(pro);// 設置優先級 
    }  

    @Override  
    public void run() {  
        for (int i = 0; i < 30; i++) {  
            System.out.println(this.getName() + "線程第" + i + "次執行！");  
            if (i % 5 == 0)  
                Thread.yield();  
        }  
    }  
}  
複製代碼

sleep方法：經過sleep(millis)使線程進入休眠一段時間，該方法在指定的時間內沒法被喚醒，同時也不會釋放對象鎖；
```
/** * 能夠明顯看到打印的數字在時間上有些許的間隔 */
public class Test1 {  
    public static void main(String[] args) throws InterruptedException {  
        for(int i=0;i<100;i++){  
            System.out.println("main"+i);  
            Thread.sleep(100);  
        }  
    }  
} 
複製代碼
```
sleep方法告訴操做系統 至少在指定時間內不需爲線程調度器爲該線程分配執行時間片，並不釋放鎖（若是當前已經持有鎖）。實際上，調用sleep方法時並不要求持有任何鎖。

因此，sleep方法並不須要持有任何形式的鎖，也就不須要包裹在synchronized中。

5 ThreadLocal

ThreadLocal是一種把變量放到線程本地的方式來實現線程同步的。好比：SimpleDateFormat不是一個線程安全的類，可使用ThreadLocal實現同步，以下：

public class ThreadLocalTest {

    private static ThreadLocal<SimpleDateFormat> dateFormatThreadLocal = new ThreadLocal<SimpleDateFormat>() {
        @Override
        protected SimpleDateFormat initialValue() {
            return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        }
    };

    public static void main(String[] args) {
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                Date date = new Date();
                System.out.println(dateFormatThreadLocal.get().format(date));
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                Date date = new Date();
                System.out.println(dateFormatThreadLocal.get().format(date));
            }
        });
        thread1.start();
        thread2.start();
    }
}
複製代碼

爲什麼SimpleDateFormat不是線程安全的類？具體請參考：

https://blog.csdn.net/zdp072/article/details/41044059

https://blog.csdn.net/zq602316498/article/details/40263083

ThreadLocal與同步機制的對比選擇：

ThreadLocal與同步機制都是 爲了解決多線程中相同變量的訪問衝突問題。

前者採用以 "空間換時間" 的方法，後者採用以 "時間換空間" 的方式。

6 volatile修飾變量

volatile關鍵字爲域變量的訪問提供了一種免鎖機制，使用volatile修飾域至關於告訴虛擬機該域可能會被其餘線程更新，所以每次使用該域就要從新計算，而不是使用寄存器中的值，volatile不會提供任何原子操做，它也不能用來修飾final類型的變量。

//只給出要修改的代碼，其他代碼與上同
public class Bank {
    //須要同步的變量加上volatile
    private volatile int account = 100;
    public int getAccount() {
        return account;
    }
    //這裏再也不須要synchronized 
    public void save(int money) {
        account += money;
    }
｝
複製代碼

多線程中的非同步問題主要出如今對域的讀寫上，若是讓域自身避免這個問題，則就不須要修改操做該域的方法。用final域，有鎖保護的域和volatile域能夠避免非同步的問題。

7 Semaphore信號量

Semaphore信號量被用於控制特定資源在同一個時間被訪問的個數。相似鏈接池的概念，保證資源能夠被合理的使用。可使用構造器初始化資源個數：

public class SemaphoreTest {

    private static Semaphore semaphore = new Semaphore(2);

    public static void main(String[] args) {
        for(int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + " " + new Date());
                        Thread.sleep(5000l);
                        semaphore.release();
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                    }
                }
            }).start();
        }
    }
}
複製代碼

輸出：

Thread-1 Mon Apr 18 18:03:46 CST 2016
Thread-0 Mon Apr 18 18:03:46 CST 2016
Thread-3 Mon Apr 18 18:03:51 CST 2016
Thread-2 Mon Apr 18 18:03:51 CST 2016
Thread-4 Mon Apr 18 18:03:56 CST 2016
複製代碼

8 併發包下的工具類

8.1 CountDownLatch

CountDownLatch是一個計數器，它的構造方法中須要設置一個數值，用來設定計數的次數。每次調用countDown()方法以後，這個計數器都會減去1，CountDownLatch會一直阻塞着調用await()方法的線程，直到計數器的值變爲0。

public class CountDownLatchTest {

    public static void main(String[] args) {
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for(int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " " + new Date() + " run");
                    try {
                        Thread.sleep(5000l);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    countDownLatch.countDown();
                }
            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("all thread over");
    }
}
複製代碼

輸出：

Thread-2 Mon Apr 18 18:18:30 CST 2016 run
Thread-3 Mon Apr 18 18:18:30 CST 2016 run
Thread-4 Mon Apr 18 18:18:30 CST 2016 run
Thread-0 Mon Apr 18 18:18:30 CST 2016 run
Thread-1 Mon Apr 18 18:18:30 CST 2016 run
all thread over
複製代碼

8.2 CyclicBarrier

CyclicBarrier阻塞調用的線程，直到條件知足時，阻塞的線程同時被打開。

調用await()方法的時候，這個線程就會被阻塞，當調用await()的線程數量到達屏障數的時候，主線程就會取消全部被阻塞線程的狀態。

在CyclicBarrier的構造方法中，還能夠設置一個barrierAction。在全部的屏障都到達以後，會啓動一個線程來運行這裏面的代碼。

public class CyclicBarrierTest {

    public static void main(String[] args) {
        Random random = new Random();
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
        for(int i = 0; i < 5; i ++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    int secs = random.nextInt(5);
                    System.out.println(Thread.currentThread().getName() + " " + new Date() + " run, sleep " + secs + " secs");
                    try {
                        Thread.sleep(secs * 1000);
                        cyclicBarrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + " " + new Date() + " runs over");
                }
            }).start();
        }
    }
}
複製代碼

相比CountDownLatch，CyclicBarrier是能夠被循環使用的，並且遇到線程中斷等狀況時，還能夠利用reset()方法，重置計數器，從這些方面來講，CyclicBarrier會比CountDownLatch更加靈活一些。

9 使用原子變量實現線程同步

有時須要使用線程同步的根本緣由在於 對普通變量的操做不是原子的。那麼什麼是原子操做呢？

原子操做就是指將讀取變量值、修改變量值、保存變量值當作一個總體來操做即-這幾種行爲要麼同時完成，要麼都不完成。

在java.util.concurrent.atomic包中提供了建立原子類型變量的工具類，使用該類能夠簡化線程同步。好比：其中AtomicInteger以原子方式更新int的值：

class Bank {
    private AtomicInteger account = new AtomicInteger(100);

    public AtomicInteger getAccount() {
        return account;
    }

    public void save(int money) {
        account.addAndGet(money);
    }
}
複製代碼

10 AbstractQueuedSynchronizer

AQS是不少同步工具類的基礎，好比：ReentrantLock裏的公平鎖和非公平鎖，Semaphore裏的公平鎖和非公平鎖，CountDownLatch裏的鎖等他們的底層都是使用AbstractQueuedSynchronizer完成的。

基於AbstractQueuedSynchronizer自定義實現一個獨佔鎖：

public class MySynchronizer extends AbstractQueuedSynchronizer {

    @Override
    protected boolean tryAcquire(int arg) {
        if(compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    @Override
    protected boolean tryRelease(int arg) {
        setState(0);
        setExclusiveOwnerThread(null);
        return true;
    }

    public void lock() {
        acquire(1);
    }

    public void unlock() {
        release(1);
    }

    public static void main(String[] args) {
        MySynchronizer mySynchronizer = new MySynchronizer();
        Thread thread1 = new Thread(new Runnable() {
            @Override
            public void run() {
                mySynchronizer.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                    System.out.println(Thread.currentThread().getName() + " will sleep 5 secs");
                    try {
                        Thread.sleep(5000l);
                        System.out.println(Thread.currentThread().getName() + " continue");
                    } catch (InterruptedException e) {
                        System.err.println(Thread.currentThread().getName() + " interrupted");
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    mySynchronizer.unlock();
                }
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            @Override
            public void run() {
                mySynchronizer.lock();
                try {
                    System.out.println(Thread.currentThread().getName() + " run");
                } finally {
                    mySynchronizer.unlock();
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}
複製代碼

11 使用阻塞隊列實現線程同步

前面幾種同步方式都是基於底層實現的線程同步，可是在實際開發當中，應當儘可能遠離底層結構。本節主要是使用LinkedBlockingQueue來實現線程的同步。

LinkedBlockingQueue是一個基於鏈表的隊列，先進先出的順序（FIFO），範圍任意的blocking queue。

package com.xhj.thread;

import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;

/** * 用阻塞隊列實現線程同步 LinkedBlockingQueue的使用 */
public class BlockingSynchronizedThread {
    /** * 定義一個阻塞隊列用來存儲生產出來的商品 */
    private LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
    /** * 定義生產商品個數 */
    private static final int size = 10;
    /** * 定義啓動線程的標誌，爲0時，啓動生產商品的線程；爲1時，啓動消費商品的線程 */
    private int flag = 0;

    private class LinkBlockThread implements Runnable {
        @Override
        public void run() {
            int new_flag = flag++;
            System.out.println("啓動線程 " + new_flag);
            if (new_flag == 0) {
                for (int i = 0; i < size; i++) {
                    int b = new Random().nextInt(255);
                    System.out.println("生產商品：" + b + "號");
                    try {
                        queue.put(b);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    System.out.println("倉庫中還有商品：" + queue.size() + "個");
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            } else {
                for (int i = 0; i < size / 2; i++) {
                    try {
                        int n = queue.take();
                        System.out.println("消費者買去了" + n + "號商品");
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    System.out.println("倉庫中還有商品：" + queue.size() + "個");
                    try {
                        Thread.sleep(100);
                    } catch (Exception e) {
                        // TODO: handle exception
                    }
                }
            }
        }
    }

    public static void main(String[] args) {
        BlockingSynchronizedThread bst = new BlockingSynchronizedThread();
        LinkBlockThread lbt = bst.new LinkBlockThread();
        Thread thread1 = new Thread(lbt);
        Thread thread2 = new Thread(lbt);
        thread1.start();
        thread2.start();
    }
}
複製代碼