JDK並發包總結

本文主要介紹jdk中常用的同步控制工具以及並發容器, 其結構如下:

同步控制工具類

ReentrantLock

簡而言之, 就是自由度更高的synchronized, 主要具備以下優點.

• 可重入: 單線程可以重復進入，但要重復退出
• 可中斷: lock.lockInterruptibly()
• 可限時: 超時不能獲得鎖，就返回false，不會永久等待構成死鎖
• 公平鎖: 先來先得, public ReentrantLock(boolean fair), 默認鎖不公平的, 根據線程優先級競爭.

示例 

public class ReenterLock implements Runnable {
    public static ReentrantLock lock = new ReentrantLock();
    public static int i = 0;

    @Override
    public void run() {
        for (int j = 0; j < 10000; j++) {
            lock.lock();
             // 超時設置
//            lock.tryLock(5, TimeUnit.SECONDS);
            try {
                i++;
            } finally {
                // 需要放在finally里釋放, 如果上面lock了兩次, 這邊也要unlock兩次
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReenterLock tl = new ReenterLock();
        Thread t1 = new Thread(tl);
        Thread t2 = new Thread(tl);
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(i);
    }
}

中斷死鎖

線程1, 線程2分別去獲取lock1, lock2, 觸發死鎖. 最終通過DeadlockChecker來觸發線程中斷.

public class DeadLock implements Runnable{

    public static ReentrantLock lock1 = new ReentrantLock();
    public static ReentrantLock lock2 = new ReentrantLock();
    int lock;

    public DeadLock(int lock) {
        this.lock = lock;
    }

    @Override
    public void run() {
        try {
            if (lock == 1){
                lock1.lockInterruptibly();
                try {
                    Thread.sleep(500);
                }catch (InterruptedException e){}
                lock2.lockInterruptibly();

            }else {
                lock2.lockInterruptibly();
                try {
                    Thread.sleep(500);
                }catch (InterruptedException e){}
                lock1.lockInterruptibly();

            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }finally {
            if (lock1.isHeldByCurrentThread())
                lock1.unlock();
            if (lock2.isHeldByCurrentThread())
                lock2.unlock();
            System.out.println(Thread.currentThread().getId() + "線程中斷");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DeadLock deadLock1 = new DeadLock(1);
        DeadLock deadLock2 = new DeadLock(2);
        // 線程1, 線程2分別去獲取lock1, lock2. 導致死鎖
        Thread t1 = new Thread(deadLock1);
        Thread t2 = new Thread(deadLock2);
        t1.start();
        t2.start();
        Thread.sleep(1000);
        // 死鎖檢查, 觸發中斷
        DeadlockChecker.check();

    }
}

public class DeadlockChecker {
    private final static ThreadMXBean mbean = ManagementFactory.getThreadMXBean();
    final static Runnable deadLockCheck = new Runnable() {
        @Override
        public void run() {
            while (true) {
                long[] deadlockedThreadlds = mbean.findDeadlockedThreads();

                if (deadlockedThreadlds != null) {
                    ThreadInfo[] threadInfos = mbean.getThreadInfo(deadlockedThreadlds);
                    for (Thread t : Thread.getAllStackTraces().keySet()) {
                        for (int i = 0; i < threadInfos.length; i++) {
                            if (t.getId() == threadInfos[i].getThreadId()) {
                                t.interrupt();
                                try {
                                    Thread.sleep(5000);
                                } catch (InterruptedException e) {
                                }
                            }
                        }
                    }
                }
            }
        }
    };

    public static void check() {
        Thread t = new Thread(deadLockCheck);
        t.setDaemon(true);
        t.start();
    }
}

Condition

類似於 Object.wait()和Object.notify(), 需要與ReentrantLock結合使用.

具體API如下:

    // await()方法會使當前線程等待，同時釋放當前鎖，當其他線程中使用signal()時或者signalAll()方法時，
    // 線程會重新獲得鎖並繼續執行。或者當線程被中斷時，也能跳出等待。這和Object.wait()方法很相似。
    void await() throws InterruptedException;
    // awaitUninterruptibly()方法與await()方法基本相同，但是它並不會再等待過程中響應中斷。
    void awaitUninterruptibly();
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    boolean awaitUntil(Date deadline) throws InterruptedException;
    // singal()方法用於喚醒一個在等待中的線程。相對的singalAll()方法會喚醒所有在等待中的線程。
    // 這和Obejct.notify()方法很類似。
    void signal();
    void signalAll();

示例

public class ReenterLockCondition implements Runnable{

    public static ReentrantLock lock = new ReentrantLock();
    public static Condition condition = lock.newCondition();

    @Override
    public void run() {
        try {
            lock.lock();
            condition.await();
            System.out.println("Thread is going on");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 注意放到finally中釋放
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReenterLockCondition t1 = new ReenterLockCondition();
        Thread tt = new Thread(t1);
        tt.start();
        Thread.sleep(2000);
        System.out.println("after sleep, signal!");
        // 通知線程tt繼續執行. 喚醒同樣需要重新獲得鎖
        lock.lock();
        condition.signal();
        lock.unlock();
    }
}

 Semaphore信號量

鎖一般都是互斥排他的, 而信號量可以認為是一個共享鎖,

允許N個線程同時進入臨界區, 但是超出許可范圍的只能等待.

如果N = 1, 則類似於lock.

具體API如下, 通過acquire獲取信號量, 通過release釋放

    public void acquire()
    public void acquireUninterruptibly()
    public boolean tryAcquire()
    public boolean tryAcquire(long timeout, TimeUnit unit)
    public void release()

示例

模擬20個線程, 但是信號量只設置了5個許可.

因此線程是按序每2秒5個的打印job done.

public class SemapDemo implements Runnable{

    // 設置5個許可
    final Semaphore semp = new Semaphore(5);

    @Override
    public void run() {
        try {
            semp.acquire();
            // 模擬線程耗時操作
            Thread.sleep(2000L);
            System.out.println("Job done! " + Thread.currentThread().getId());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semp.release();
        }
    }

    public static void main(String[] args){
        ExecutorService service = Executors.newFixedThreadPool(20);
        final SemapDemo demo = new SemapDemo();
        for (int i = 0; i < 20; i++) {
            service.submit(demo);
        }
    }
}

ReadWriteLock

讀寫分離鎖, 可以大幅提升系統並行度.

• 讀-讀不互斥：讀讀之間不阻塞。
• 讀-寫互斥：讀阻塞寫，寫也會阻塞讀。
• 寫-寫互斥：寫寫阻塞。

示例

使用方法與ReentrantLock類似, 只是讀寫鎖分離.

private static ReentrantReadWriteLock readWriteLock=new ReentrantReadWriteLock();
private static Lock readLock = readWriteLock.readLock();
private static Lock writeLock = readWriteLock.writeLock();

CountDownLatch倒數計時器

一種典型的場景就是火箭發射。在火箭發射前，為了保證萬無一失，往往還要進行各項設備、儀器的檢查。

只有等所有檢查完畢后，引擎才能點火。這種場景就非常適合使用CountDownLatch。它可以使得點火線程, 

等待所有檢查線程全部完工后，再執行.

示例

public class CountDownLatchDemo implements Runnable{
    static final CountDownLatch end = new CountDownLatch(10);
    static final CountDownLatchDemo demo = new CountDownLatchDemo();

    @Override
    public void run() {
        try {
            Thread.sleep(new Random().nextInt(10) * 1000);
            System.out.println("check complete!");
            end.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            service.submit(demo);
        }
        // 等待檢查
        end.await();
        // 所有線程檢查完畢, 發射火箭.
        System.out.println("fire");
        service.shutdown();
    }
}

CyclicBarrier循環柵欄

Cyclic意為循環，也就是說這個計數器可以反復使用。比如，假設我們將計數器設置為10。那么湊齊

第一批10個線程后，計數器就會歸零，然后接着湊齊下一批10個線程.

示例

public class CyclicBarrierDemo {

    public static class Soldier implements Runnable {

        private String soldier;
        private final CyclicBarrier cyclic;

        Soldier(CyclicBarrier cyclic, String soldier) {
            this.cyclic = cyclic;
            this.soldier = soldier;
        }

        @Override
        public void run() {
            try {
                // 等待所有士兵到期
                cyclic.await();
                doWork();
                // 等待所有士兵完成工作
                cyclic.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }

        void doWork() {
            try {
                Thread.sleep(Math.abs(new Random().nextInt() % 10000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(soldier + " 任務完成!");
        }
    }

    public static class BarrierRun implements Runnable {
        boolean flag;
        int N;

        public BarrierRun(boolean flag, int n) {
            this.flag = flag;
            N = n;
        }

        @Override
        public void run() {
            if (flag) {
                System.out.println("士兵:" + N + "個, 任務完成!");
            } else {
                System.out.println("士兵:" + N + "個, 集合完畢!");
                flag = true;
            }
        }
    }

    public static void main(String[] args){
        final int N = 5;
        Thread[] allSoldier = new Thread[N];
        boolean flag = false;
        CyclicBarrier cyclic = new CyclicBarrier(N, new BarrierRun(flag, N));
        // 設置屏障點, 主要為了執行這個方法.
        System.out.println("集合任務!");
        for (int i = 0; i < N; i++) {
            System.out.println("士兵" + i + " 報到!");
            allSoldier[i] = new Thread(new Soldier(cyclic, "士兵" + i));
            allSoldier[i].start();
        }

    }
} 

結果

集合任務!
士兵0 報到!
士兵1 報到!
士兵2 報到!
士兵3 報到!
士兵4 報到!
士兵:5個, 集合完畢!
士兵3 任務完成!
士兵1 任務完成!
士兵0 任務完成!
士兵4 任務完成!
士兵2 任務完成!
士兵:5個, 任務完成!

LockSupport

一個線程阻塞工具, 可以在任意位置讓線程阻塞.

與suspend()比較, 如果unpark發生在park之前, 並不會導致線程凍結, 也不需要獲取鎖.

API

LockSupport.park();
LockSupport.unpark(t1);

中斷響應

能夠響應中斷，但不拋出異常。

中斷響應的結果是，park()函數的返回，可以從Thread.interrupted()得到中斷標志

public class LockSupportDemo {
    public static Object u = new Object();
    static ChangeObjectThread t1 = new ChangeObjectThread("t1");
    static ChangeObjectThread t2 = new ChangeObjectThread("t2");
    public static class ChangeObjectThread extends Thread {

        public ChangeObjectThread(String name) {
            super(name);
        }

        @Override
        public void run() {
            synchronized (u) {
                System.out.println("in " + getName());
                LockSupport.park();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        t1.start();
        Thread.sleep(100);
        t2.start();
        LockSupport.unpark(t1);
        LockSupport.unpark(t2);
        t1.join();
        t2.join();
    }
}

並發容器

Collections.synchronizedMap

其本質是在讀寫map操作上都加了鎖, 因此不推薦在高並發場景使用.

ConcurrentHashMap

內部使用分區Segment來表示不同的部分, 每個分區其實就是一個小的hashtable. 各自有自己的鎖.

只要多個修改發生在不同的分區, 他們就可以並發的進行. 把一個整體分成了16個Segment, 最高支持16個線程並發修改.

代碼中運用了很多volatile聲明共享變量, 第一時間獲取修改的內容, 性能較好.

    public V put(K key, V value) {
        ConcurrentHashMap.Segment<K,V> s;
        if (value == null)
            throw new NullPointerException();
        int hash = hash(key);
        int j = (hash >>> segmentShift) & segmentMask;
        // 通過unsafe對j進行偏移來尋找key所對應的分區
        if ((s = (ConcurrentHashMap.Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
                (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
            // 如果分區不存在, 則創建新的分區
            s = ensureSegment(j);
        // kv放到分區中
        return s.put(key, hash, value, false);
    }

Segment.put源碼

    Segment(float lf, int threshold, ConcurrentHashMap.HashEntry<K,V>[] tab) {
        this.loadFactor = lf;
        this.threshold = threshold;
        this.table = tab;
    }

    final V put(K key, int hash, V value, boolean onlyIfAbsent) {
        // tryLock通過無鎖cas操作嘗試獲取鎖(無等待), 繼承自ReentrantLock.
        // 如果成功則, node = null
        // 如果不成功, 則可能其他線程已經在插入數據了,
        // 此時會嘗試繼續獲取鎖tryLock, 自旋MAX_SCAN_RETRIES次, 若還是拿不到鎖才開始lock
        ConcurrentHashMap.HashEntry<K,V> node = tryLock() ? null :
                scanAndLockForPut(key, hash, value);
        V oldValue;
        try {
            ConcurrentHashMap.HashEntry<K,V>[] tab = table;
            // 獲取分區中哪一個entry鏈的index
            int index = (tab.length - 1) & hash;
            // 獲取第一個entry
            ConcurrentHashMap.HashEntry<K,V> first = entryAt(tab, index);
            for (ConcurrentHashMap.HashEntry<K,V> e = first;;) {
                // e != null , 存在hash沖突, 把他加到當前鏈表中
                if (e != null) {
                    K k;
                    if ((k = e.key) == key ||
                            (e.hash == hash && key.equals(k))) {
                        oldValue = e.value;
                        if (!onlyIfAbsent) {
                            e.value = value;
                            ++modCount;
                        }
                        break;
                    }
                    e = e.next;
                }
                else {
                    // 無hash沖突, new entry
                    if (node != null)
                        node.setNext(first);
                    else
                        node = new ConcurrentHashMap.HashEntry<K,V>(hash, key, value, first);
                    int c = count + 1;
                    // 空間大小超出閾值, 需要rehash, 翻倍空間.
                    if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                        rehash(node);
                    else
                        //放到分區中
                        setEntryAt(tab, index, node);
                    ++modCount;
                    count = c;
                    oldValue = null;
                    break;
                }
            }
        } finally {
            unlock();
        }
        return oldValue;
    }

如果想要對ConcurrentHashMap排序, 則可以使用ConcurrentSkipListMap,

他支持並發排序, 是一個線程安全的類似TreeMap的實現.

BlockingQueue

阻塞隊列, 主要用於多線程之間共享數據.

當一個線程讀取數據時, 如果隊列是空的, 則當前線程會進入等待狀態.

如果隊列滿了, 當一個線程嘗試寫入數據時, 同樣會進入等待狀態.

適用於生產消費者模型.

其源碼實現也相對簡單.

    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 隊列滿了, 寫進入等待
            while (count == items.length)
                notFull.await();
            insert(e);
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 隊列空的, 讀進入等待
            while (count == 0)
                notEmpty.await();
            return extract();
        } finally {
            lock.unlock();
        }
    }

因為BlockingQueue在put take等操作有鎖, 因此非高性能容器, 

如果需要高並發支持的隊列, 則可以使用ConcurrentLinkedQueue. 他內部也是運用了大量無鎖操作.

CopyOnWriteArrayList

CopyOnWriteArrayList通過在新增元素時, 復制一份新的數組出來, 並在其中寫入數據, 之后將原數組引用指向到新數組.

其Add操作是在內部通過ReentrantLock進行鎖保護, 防止多線程場景復制多份數組.

而Read操作內部無鎖, 直接返回數組引用, 並發下效率高, 因此適用於讀多寫少的場景.

源碼

    public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        // 寫數據的鎖
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            // 復制到新的數組
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            // 加入新元素
            newElements[len] = e;
            // 修改引用
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }

    final void setArray(Object[] a) {
        array = a;
    }

    // 讀的時候無鎖
    public E get(int index) {
        return get(getArray(), index);
    }

示例

使用10個讀線程, 100個寫線程. 如果使用ArrayList實現, 那么有可能是在運行過程中拋出ConcurrentModificationException.

原因很簡單, ArrayList在遍歷的時候會check modCount是否發生變化, 如果一邊讀一邊寫就會拋異常.

public class CopyOnWriteListDemo {

    static List<UUID> list = new CopyOnWriteArrayList<UUID>();
//    static List<UUID> list = new ArrayList<UUID>();

    // 往list中寫數據
    public static class AddThread implements Runnable {

        @Override
        public void run() {
            UUID uuid = UUID.randomUUID();
            list.add(uuid);
            System.out.println("++Add uuid : " + uuid);

        }
    }

    // 從list中讀數據
    public static class ReadThread implements Runnable {

        @Override
        public void run() {
            System.out.println("start read size: " + list.size() + " thread : " + Thread.currentThread().getName());
            for (UUID uuid : list) {
                System.out.println("Read uuid : " + uuid + " size : " + list.size() + "thread: " + Thread.currentThread().getName());
            }
        }
    }


    public static void main(String[] args) throws InterruptedException {
        initThread(new AddThread(), 10);
        initThread(new ReadThread(), 100);
    }

    private static void initThread(Runnable runnable, int maxNum) throws InterruptedException {
        Thread[] ts = new Thread[maxNum];
        for (int k = 0; k < maxNum; k++) {
            ts[k] = new Thread(runnable);
        }
        for (int k = 0; k < maxNum; k++) {
            ts[k].start();
        }
    }
}

下圖運行結果中可以看出來, 同一個線程, 即使在讀的過程中發生了size變化, 也不會拋出ConcurrentModificationException