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