Java併發編程之線程同步

時間 2019-12-05

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

原文原文鏈接

線程安全就是防止某個對象或者值在多個線程中被修改而致使的數據不一致問題，所以咱們就須要經過同步機制保證在同一時刻只有一個線程可以訪問到該對象或數據，修改數據完畢以後，再將最新數據同步到主存中，使得其餘線程都可以獲得這個最新數據。下面咱們就來了解Java一些基本的同步機制。java

volatile關鍵字

Java提供了一種稍弱的同步機制即volatile變量，用來確保將變量的更新操做通知到其餘線程。當把變量聲明爲volatile類型後，編譯器與運行時都會注意到這個變量是共享的。然而，在訪問volatile變量時不會執行加鎖操做，所以也就不會使線程阻塞，所以volatile變量是一種比synchronized關鍵字更輕量級的同步機制。數組

volatile變量對全部的線程都是可見的，對volatile變量全部的寫操做都能當即反應到其餘線程之中，即volatile變量在各個線程中是一致的。安全

有一種狀況須要注意：volatile的語義不能確保遞增（count++）的原子性，除非你能確保只有一個線程對變量執行寫操做。async

public class VolatileTest{
    
    public static volatile int  i;

    public static void increase(){
        i++;
    }
}

查看字節碼: javap -c -l VolatileTest.class

public class VolatileTest {
  public static volatile int i;
 
  public VolatileTest();
    Code:
       0: aload_0       
       1: invokespecial #1                  // Method java/lang/Object."<init>":()V
       4: return        
    LineNumberTable:
      line 1: 0
 
  public static void increase();
    Code:
       0: getstatic     #2                  // Field i:I, 把i的值取到了操做棧頂,volatile保證了i值此時是正確的. 
       3: iconst_1      
       4: iadd                              // increase,但其餘線程此時可能已經把i值加大了好多
       5: putstatic     #2                  // Field i:I ,把這個已經out of date的i值同步回主內存中,i值被破壞了.
       8: return        
    LineNumberTable:
      line 6: 0
      line 7: 8
}

加鎖機制便可以確保原子性又能夠確保可見性，而volatile變量只能確保可見性。ide

內置鎖-synchronized

Java中最經常使用的同步機制就是synchronized關鍵字，它是一種基於語言的粗略鎖，可以做用於對象、函數、Class。每一個對象都只有一個鎖，誰可以拿到這個鎖誰就獲得了訪問權限。當synchronized做用於函數時，實際上鎖的也是對象，鎖定的對象是該函數所在類的對象。而synchronized做用於Class時則鎖的是這個Class類，並不是某個具體對象。函數

synchronized同步方法和同步塊

public class SynchronizedDemo {
    
    
    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        final Test test = new Test();

        new Thread(new Runnable() {
            
            @Override
            public void run() {
                // TODO Auto-generated method stub
                test.syncMethod(Thread.currentThread());
                
            }
        }).start();
        
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                // TODO Auto-generated method stub
                test.syncMethod(Thread.currentThread());
            }
        }).start();
        
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                // TODO Auto-generated method stub
                test.asyncMethod(Thread.currentThread());
            }
        }).start();
        
    }
}

class Test {
    
    public synchronized void syncMethod(Thread thread) {
        for(int i = 0;i < 3;i++) {
            System.out.println(thread.getName());
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    
    public void asyncMethod(Thread thread) {
        
        synchronized (this) {
            
            for(int i = 0;i < 3;i++) {
                System.out.println(thread.getName()+2);
            }
        }
    }
}

syncMethod和asyncMethod代碼塊都加鎖時結果:

Thread-0
Thread-0
Thread-0
Thread-1
Thread-1
Thread-1
Thread-2
Thread-2
Thread-2  #多個線程不能同時訪問同一個對象中的synchronized鎖的方法或代碼塊

syncMethod加鎖和asyncMethod代碼塊不加鎖時結果:

class Test {
    
    public synchronized void syncMethod(Thread thread) {
        for(int i = 0;i < 3;i++) {
            System.out.println(thread.getName());
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    
    public void asyncMethod(Thread thread) {
        
        synchronized (this) {
            
            for(int i = 0;i < 3;i++) {
                System.out.println(thread.getName());
            }
        }
    }
}

Thread-0
Thread-22
Thread-22
Thread-22
Thread-0
Thread-0
Thread-1
Thread-1
Thread-1  #其餘線程能夠訪問同一個對象的非同步方法或代碼塊

syncMethod不加鎖和asyncMethod代碼塊不加鎖時結果:

class Test {
    
    public void syncMethod(Thread thread) {
        for(int i = 0;i < 3;i++) {
            System.out.println(thread.getName());
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    
    public void asyncMethod(Thread thread) {
        
            for(int i = 0;i < 3;i++) {
                System.out.println(thread.getName()+2);
            }
    }
}

Thread-0
Thread-1
Thread-22
Thread-22
Thread-22
Thread-0
Thread-1
Thread-1
Thread-0

synchronized同步方法和同步塊鎖定的是引用對象，synchronized做用於引用對象是防止其餘線程訪問同一個對象的synchronized代碼塊或方法，但能夠訪問其餘非同步代碼塊或方法。ui

synchronized同步Class對象和靜態方法

public class SynchronizedDemo {

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        final Test test = new Test();

        new Thread(new Runnable() {

            @Override
            public void run() {
                // TODO Auto-generated method stub
                Test.syncStaticMethod(Thread.currentThread());

            }
        }).start();

        new Thread(new Runnable() {

            @Override
            public void run() {
                // TODO Auto-generated method stub
                Test.syncStaticMethod(Thread.currentThread());
            }
        }).start();

        new Thread(new Runnable() {

            @Override
            public void run() {
                // TODO Auto-generated method stub
                Test.asyncStaticMethod(Thread.currentThread());
            }
        }).start();

    }
}

class Test {

    public synchronized static void syncStaticMethod(Thread thread) {

        for (int i = 0; i < 3; i++) {
            System.out.println(thread.getName());
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

    public static void asyncStaticMethod(Thread thread) {

        synchronized (Test.class) {
            for (int i = 0; i < 3; i++) {
                System.out.println(thread.getName() + 22);
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }

    }
}
syncStaticMethod和asyncStaticMethod代碼塊都加鎖的結果:

Thread-0
Thread-0
Thread-0
Thread-222
Thread-222
Thread-222
Thread-1
Thread-1
Thread-1  ##多個線程不能同時訪問添加了synchronized鎖的代碼塊和方法。

syncStaticMethod加鎖和asyncStaticMethod代碼塊不加鎖的結果:

class Test {

    public synchronized static void syncStaticMethod(Thread thread) {

        for (int i = 0; i < 3; i++) {
            System.out.println(thread.getName());
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

    public static void asyncStaticMethod(Thread thread) {

            for (int i = 0; i < 3; i++) {
                System.out.println(thread.getName() + 22);
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
    }
}

Thread-0
Thread-222
Thread-222
Thread-0
Thread-0
Thread-222
Thread-1
Thread-1
Thread-1 ##多個線程能夠同時訪問非同步的代碼塊和方法

syncStaticMethod加鎖和asyncStaticMethod代碼塊都不加鎖的結果:

class Test {

    public static void syncStaticMethod(Thread thread) {

        for (int i = 0; i < 3; i++) {
            System.out.println(thread.getName());
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

    public static void asyncStaticMethod(Thread thread) {

            for (int i = 0; i < 3; i++) {
                System.out.println(thread.getName() + 22);
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
    }
}

Thread-0
Thread-1
Thread-222
Thread-1
Thread-0
Thread-222
Thread-1
Thread-0
Thread-222

synchronized同步Class對象和靜態方法鎖的是Class對象，它的做用是防止多個線程同時訪問添加了synchronized鎖的代碼塊和方法。this

總結

當一個線程正在訪問一個對象的synchronized方法，那麼其餘線程不能訪問該對象的其餘synchronized方法，由於一個對象只有一把鎖，當一個線程獲取了該對象的鎖以後，其餘線程沒法獲取該對象的鎖，全部沒法訪問該對象的其餘synchronized方法。線程
當一個線程正在訪問一個對象的synchronized方法，那麼其餘線程能訪問該對象的非synchronized方法。由於非synchronized方法不須要獲取該對象的鎖。code
若是一個線程A須要訪問對象object1的synchronized方法fun1，另一個線程B須要訪問對象object2的synchronized方法fun1，即便object1和object2是同一類型，也不會產生線程安全問題，由於他們訪問的是不一樣的對象，因此不存在互斥問題。
若是一個線程執行一個對象的非static synchronized方法，另外一個線程執行這個對象所屬類的static synchronized方法，此時不會發生互斥現象，由於訪問static synchronized方法佔用的是類鎖，而訪問非static synchronized方法佔用的是對象鎖，因此不存在互斥現象。

須要注意的是：對於synchronized方法或者synchronized代碼塊，當出現異常時，JVM會自動釋放當前線程佔用的鎖，所以不會因爲異常致使出現死鎖現象。

顯示鎖-ReentrantLock與Condition

ReentrantLock

在JDk 5.0以前，協調共享對象的訪問時，只有synchronized和volatile。Java 6.0增長了一種新的機制：ReentrantLock。顯示鎖ReentrantLock和內置鎖synchronized相比，實現了相同的語義，可是具備更高的靈活性。

內置鎖synchronized的獲取和釋放都在同一個代碼塊中，而顯示鎖ReentrantLock則能夠將鎖的得到和釋放分開。同時顯示鎖能夠提供輪訓鎖和定時鎖，同時能夠提供公平鎖或者非公平鎖。

ReentrantLock的基本操做以下：

函數	做用
lock()	獲取鎖
tryLock()	嘗試獲取鎖
tryLock(timeout,Timeunit unit)	在指定時間內嘗試獲取鎖
unLock()	釋放鎖
newCondition	獲取鎖的Condition

使用ReentrantLock的通常是lock、tryLock與unLock成對出現，須要注意的是，千萬不要忘記調用unLock來釋放鎖，不然會引起死鎖等問題。

ReentrantLock的經常使用形式以下所示：

Lock lock = new ReentrantLock();

public void run() {

    lock.lock();

    try {

        //執行任務

    } finally {

        lock.unlock();
    }
}

須要注意的是，lock必須在finally塊中釋放，不然，若是受保護的代碼塊拋出異常，鎖就有可能永遠得不到釋放。而使用synchronized同步，JVM將確保鎖會得到自動釋放，這也是Lock沒有徹底替代掉synchronized的緣由。

當JVM用synchronized管理鎖定請求和釋放行爲時，JVM在生成線程轉儲時可以包括鎖定信息，這些對調式有很是大的價值，由於它們能標識死鎖和其餘異常行爲的來源。Lock類只是普通的類，JVM不知道具體哪一個線程擁有Lock對象。

Condition

在ReentrantLock類中有一個重要的函數newCondition()，該函數用於獲取lock上的一個條件，也就是說Condition是和Lock綁定的。Condition用於實現線程間的通訊，它是爲了解決Object.wait()、notify()、notifyAll()難以使用的問題。

Condition的基本操做以下所示：

方法	做用
await()	線程等待
await(int time,TimeUnit unit)	線程等待特定的時間，超過期間則爲超時
signal()	隨機喚醒某個等待線程
signalAll()	喚醒全部等待中的線程

綜合應用

下面經過ReentrantLock和Condition類實現一個簡單的阻塞隊列。若是調用take方法時集合中沒有數據，那麼調用線程阻塞；若是調用put方法時，集合數據已滿則調用線程阻塞。可是這兩個阻塞條件是不一樣的，分別爲notFull和notEmpty。MyArrayBlockingQueue的實現代碼以下：

public class MyArrayBlockingQueue<T> {

    // 數據數組
    private final T[] items;
    // 鎖
    private final Lock mLock = new ReentrantLock();
    // 數組滿的條件
    private Condition notFull = mLock.newCondition();
    // 數組空的條件
    private Condition notEmpty = mLock.newCondition();

    // 頭部
    private int head;
    // 尾部
    private int tail;
    // 數據數量
    private int count;

    public MyArrayBlockingQueue(int maxSize) {
        // TODO Auto-generated constructor stub
        items = (T[]) new Object[maxSize];
    }

    public MyArrayBlockingQueue() {
        // TODO Auto-generated constructor stub
        this(10);
    }

    public void put(T t) {

        mLock.lock();

        try {

            // 若是數據已滿,等待
            while (count == getCapacity()) {
                System.out.println("數據已滿，請等待");
                notFull.await();
            }

            System.out.println("存入數據");

            items[tail] = t;
            if (++tail == getCapacity()) {
                tail = 0;
            }

            ++count;
            // 喚醒等待數據的線程
            notEmpty.signalAll();

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            mLock.unlock();
        }
    }

    public T take() {

        mLock.lock();

        try {

            // 若是數組數據爲空，則阻塞
            while (count == 0) {
                System.out.println("尚未數據，等待");
                notEmpty.await();
            }

            System.out.println("取出數據");

            T t = items[head];
            items[head] = null;

            if (++head == getCapacity()) {
                head = 0;
            }

            --count;
            // 喚醒添加數據的線程
            notFull.signalAll();
            return t;

        } catch (InterruptedException e) {
            // TODO: handle exception
        } finally {
            mLock.unlock();
        }

        return null;
    }

    public int getCapacity() {
        return items.length;
    }

    public int size() {
        mLock.lock();
        try {
            return count;
        } finally {
            mLock.unlock();
        }
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        final MyArrayBlockingQueue<String> mQueue = new MyArrayBlockingQueue<>(
                5);

        new Thread(new Runnable() {

            @Override
            public void run() {
                // TODO Auto-generated method stub
                while (true) {

                    for(int i  = 0;i < 3;i++)
                    mQueue.put("just");
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }

                }

            }
        }).start();

        new Thread(new Runnable() {

            @Override
            public void run() {
                // TODO Auto-generated method stub
                while (true) {
                    
                    mQueue.take();

                }
            }
        }).start();

    }

}

結果打印
存入數據
存入數據
存入數據
取出數據
取出數據
取出數據
尚未數據，等待
存入數據
存入數據
存入數據
取出數據
取出數據
取出數據
尚未數據，等待

信號量-Semaphore

Semaphore是一個計數信號量，它的本質是一個「共享鎖」。信號量維護一個信號許可集合，線程能夠經過調用acquire()來獲取信號量的許可。當信號量有可用的許可時，線程能獲取該許可；不然線程必須等到，直到有可用的許可爲止。線程能夠經過release()來釋放它所持有的信號量許可。

Semaphore實現的功能相似食堂窗口。例如，食堂只有3個銷售窗口，要吃飯的有5我的，那麼同時只有3我的買飯菜，每一個人佔用一個窗口，另外2人只能等待。當前3我的有人離開以後，後續的人才能夠佔用窗口進行購買。這裏的窗口就是咱們所說的許可集，這裏爲3.一我的佔用窗口時至關於他調用acquire()獲取了許可，當他離開時也就等於調用release()釋放了許可，這樣後續的人才能夠獲得許可。下面看看具體的示例：

public class SemaphoreTest {
    

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        ExecutorService service = Executors.newFixedThreadPool(3);
        final Semaphore semaphore = new Semaphore(3);
        
        for(int i = 0;i < 5;i++) {
            
            service.submit(new Runnable() {
                
                @Override
                public void run() {
                    // TODO Auto-generated method stub
                    try {
                        semaphore.acquire();
                        System.out.println("剩餘許可: " + semaphore.availablePermits());
                        Thread.sleep(2000);
                        semaphore.release();
                        
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    
                }
            });
        }

    }

}

結果打印：

剩餘許可: 0
剩餘許可: 0
剩餘許可: 0

剩餘許可: 2
剩餘許可: 1

上述結果中：前三行是馬上輸出的，後兩行是等待2秒以後才輸出。緣由是，信號量的許可集是3個，而消費線程是5個。前3個線程獲取了許可以後，信號量的許可就爲0。此時後面的線程再調用acquire()就會阻塞，直到前3個線程執行完以後，釋放了許可（不須要同時釋放許可）後兩個線程才能獲取許可而且繼續執行。

循環柵欄-CyclicBarrier

CyclicBarrier是一個同步輔助類，容許一組線程互相等待，直到達到某個公共屏障點。由於該barrier在釋放等待線程後能夠重用，全部稱爲循環的barrier。

下面看看示例：

public class CyclicBarrierTest {
    
    private static final int SIZE = 5;
    private static CyclicBarrier mCyclicBarrier;

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub

        mCyclicBarrier = new CyclicBarrier(SIZE, new Runnable() {
            
            @Override
            public void run() {
                // TODO Auto-generated method stub
                System.out.println("--知足條件執行特定操做，參與者： "+ mCyclicBarrier.getParties());
            }
        });
        
        
        for(int i = 0;i < SIZE;i++) {
            new WorkerThread().start();
        }
        
        
    }
    
    static class WorkerThread extends Thread {
        
        @Override
        public void run() {
            // TODO Auto-generated method stub
            try {
                System.out.println(Thread.currentThread().getName() + "等待CyclicBarrier");
                //將mCyclicBarrier的參與者數量加1
                mCyclicBarrier.await();
                //mCyclicBarrier的參與者數量加5時，才繼續日後執行
                System.out.println(Thread.currentThread().getName()+"繼續執行");
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

}

結果打印：

Thread-1等待CyclicBarrier
Thread-0等待CyclicBarrier
Thread-2等待CyclicBarrier
Thread-3等待CyclicBarrier
Thread-4等待CyclicBarrier
--知足條件執行特定操做，參與者： 5
Thread-4繼續執行
Thread-3繼續執行
Thread-2繼續執行
Thread-0繼續執行
Thread-1繼續執行

從結果能夠看出，只有當有5個線程調用了mCyclicBarrier.await()方法後，後續的任務纔會繼續執行。上述例子中的5個WorkThread就位以後首先會執行一個Runnable，也就是CyclicBarrier構造函數的第二個參數，該參數也能夠省略。執行該Runnable以後纔會繼續執行下面的任務。CyclicBarrier實際上至關於能夠用於多個線程等待，直到某個條件被知足後開始繼續執行後續的任務。對於該示例來講，這裏的條件也就是有指定個數的線程調用了mCyclicBarrier.await()方法。

閉鎖-CountDownLatch

CountDownLatch是一個同步輔助類，在完成一組正在其餘線程中執行的操做以前，它容許一個或多個線程一直等待，直到條件被知足。

示例以下：

public class CountDownLatchTest {
    
    private static final int LATCH_SIZE = 5;

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        try {
            
            CountDownLatch countDownLatch = new CountDownLatch(LATCH_SIZE);
            
            for(int i = 0;i < LATCH_SIZE;i++) {
                new WorkerThread(countDownLatch).start();
            }
    
            System.out.println("主線程等待");
            
            countDownLatch.await();
            
            System.out.println("主線程繼續執行");
            
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    
    static class WorkerThread extends Thread {
        
        private CountDownLatch latch;
        
        public WorkerThread(CountDownLatch latch) {
            this.latch = latch;
        }
        
        @Override
        public void run() {
            // TODO Auto-generated method stub
            try {
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() + "執行操做");
                //將latch的數量減1
                latch.countDown();
                
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

}

結果打印：

主線程等待
Thread-3執行操做
Thread-1執行操做
Thread-0執行操做
Thread-4執行操做
Thread-2執行操做
主線程繼續執行

5個WorkThread對象在執行完操做以後會調用CountDownLatch的countDown()函數，當5個WorkThread全都調用了countDown()以後主線程就會被喚醒繼續執行任務。