Java篇 - 锁机制与应用场景全集3 (阻塞队列, 死锁, CountdownLatch, CyclicBarrier)

今天更新java中的锁机制：第三章（大纲9-12）。

终于要把java锁这块收尾了，不容易。下个博文的主题是：Java并发容器类源码分析和性能对比，这章我本来想放在锁机制里的，但是感觉并发容器可以单独抽出一篇，主要分析源码及性能对比。

大纲：

1. 并发的特性
2. 锁的分类
3. synchronized
4. volatile
5. Lock
6. ThreadLocal
7. Atmoic
8. Semaphore
9. 阻塞队列
10. 死锁
11. CountdownLatch
12.CyclicBarrier

9. 阻塞队列

java中提供了很多非阻塞队列，比如PriorityQueue、LinkedList（LinkedList是双向链表，它实现了Dequeue接口）。

使用非阻塞队列有一个问题，就是它不会对当前线程造成阻塞。但是我们前面讲的生产者-消费者模型，如果使用非阻塞队列，就需要额外的实现同步策略和线程间唤醒策略。如果使用阻塞队列，它会对当前线程产生阻塞，比如一个线程从一个空的阻塞队列中取元素，此时线程会被阻塞直到阻塞队列中有了元素。当队列中有元素后，被阻塞的线程会自动被唤醒（不需要我们编写代码去唤醒），这样提供了极大的方便性。

9.1 几种主要的阻塞队列

从JDK1.5之后，java就提供了几种阻塞队列，位于java.util.concurrent包。

ArrayBlockingQueue：基于数组实现的一个阻塞队列，在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性，默认情况下为非公平的，即不保证等待时间最长的队列最优先能够访问队列。

LinkedBlockingQueue：基于链表实现的一个阻塞队列，在创建LinkedBlockingQueue对象时如果不指定容量大小，则默认大小为Integer.MAX_VALUE。

PriorityBlockingQueue：以上2种队列都是先进先出队列，而PriorityBlockingQueue却不是，它会按照元素的优先级对元素进行排序，按照优先级顺序出队，每次出队的元素都是优先级最高的元素。注意，此阻塞队列为无界阻塞队列，即容量没有上限（通过源码就可以知道，它没有容器满的信号标志），前面2种都是有界队列。

DelayQueue：基于PriorityQueue，一种延时阻塞队列，DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue也是一个无界队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。

9.2 非阻塞队列和阻塞队列方法对比

非阻塞队列中的几个主要方法：

add(E e): 将元素e插入到队列末尾，如果插入成功，则返回true，如果插入失败（即队列已满），则会抛出异常；

remove()：移除队首元素，若移除成功，则返回true，如果移除失败（队列为空），则会抛出异常；

offer(E e)：将元素e插入到队列末尾，如果插入成功，则返回true，如果插入失败（即队列已满），则返回false；

poll()：移除并获取队首元素，若成功，则返回队首元素，并移除队首，否则返回null；

peek()：获取队首元素，若成功，则返回队首元素，否则返回null；

对于非阻塞队列，一般情况下建议使用offer、poll和peek三个方法，不建议使用add和remove方法。因为使用offer、poll和peek三个方法可以通过返回值判断操作成功与否，而使用add和remove方法却不能达到这样的效果，而且需要自己处理异常。注意，非阻塞队列中的方法都没有进行同步措施。

阻塞队列中的几个主要方法：

put(E e): 用来向队尾存入元素，如果队列满，则等待；

take(): 用来从队首取元素，如果队列为空，则等待；

offer(E e): 用来向队尾存入元素，如果队列满返回false，否则返回true;

offer(E e,long timeout, TimeUnit unit): 用来向队尾存入元素，如果队列满，则等待一定的时间，当时间期限达到时，如果还没有插入成功，则返回false，否则返回true；

poll(): 用来从队首取元素，如果队列空则返回null，否则返回取得的元素，并移除队首；

poll(long timeout, TimeUnit unit): 用来从队首取元素，如果队列空，则等待一定的时间，当时间期限达到时，如果取到，则返回null，否则返回取得的元素，并移除队首；

peek(): 用来从队首取元素，如果队列为空，则返回null，否则返回取得的元素；

9.3 阻塞队列的实现原理

以ArrayBlockingQueue为例：

/** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;
    
    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

上面是ArrayBlockingQueue的构造器，可以看到使用了重入锁ReentrantLock以及Condition实现同步策略和线程间唤醒策略。

public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
   }

   public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

   public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {

        checkNotNull(e);
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length) {
                if (nanos <= 0)
                    return false;
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(e);
            return true;
        } finally {
            lock.unlock();
        }
    }

    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
    }

    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return itemAt(takeIndex); // null when queue is empty
        } finally {
            lock.unlock();
        }
    }

add(E e)方法：内部调用了offer方法，如果队列已满则抛出异常。

offer(E e)方法: 插入队尾的时候加锁，如果队列已满返回false，否则插入并返回true。

offer(E e, long timeout, TimeUnit unit): 插入队尾，先加锁lockInterruptibly，如果在等待锁的过程中，可以调用interrupt()中断其等待。如果队列已满，则notFull.awaitNanos(nanos)，如果在这期间队列有位置，则插入并返回true，否则返回false。

put(E e): 插入队尾的时候加锁，如果队列已满，则等待notFull.await()，直到队列有空余位置。

poll(): 从队首取元素时加锁，如果队列为空，返回null，否则返回并移除队首元素。

poll(long timeout, TimeUnit unit) : 从队首取元素时加锁，如果队列为空，则等待notEmpty.awaitNanos(nanos)，如果这期间队列有元素则返回并移除队首元素，否则返回null。

take(): 从队首取元素时加锁，如果队列为空，则等待notEmpty.await()，直到队列中有元素，移除队首并返回。

peek(): 从队尾取元素时加锁，如果队列为空返回null，否则返回队首元素。

9.4 阻塞队列使用例子

private static final int QUEUE_SIZE = 12;

    private static final class Consumer extends Thread {

        private final BlockingQueue<Integer> queue;

        Consumer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            consume();
        }

        private void consume() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    queue.take();
                    System.out.println("从队列取走一个元素，队列剩余" + queue.size() + "个元素");
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static final class Producer extends Thread {

        private final BlockingQueue<Integer> queue;

        Producer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            produce();
        }

        private void produce() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    queue.put(1);
                    System.out.println("向队列取中插入一个元素，队列剩余空间：" + (QUEUE_SIZE - queue.size()));
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(QUEUE_SIZE);
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        producer.start();
        consumer.start();
    }

执行输出：

向队列取中插入一个元素，队列剩余空间：12

从队列取走一个元素，队列剩余0个元素

向队列取中插入一个元素，队列剩余空间：11

从队列取走一个元素，队列剩余0个元素

向队列取中插入一个元素，队列剩余空间：11

从队列取走一个元素，队列剩余0个元素

向队列取中插入一个元素，队列剩余空间：11

从队列取走一个元素，队列剩余0个元素

从队列取走一个元素，队列剩余0个元素

向队列取中插入一个元素，队列剩余空间：12

向队列取中插入一个元素，队列剩余空间：12

从队列取走一个元素，队列剩余0个元素

向队列取中插入一个元素，队列剩余空间：11

从队列取走一个元素，队列剩余0个元素

向队列取中插入一个元素，队列剩余空间：11

从队列取走一个元素，队列剩余0个元素

可以看到，生产者消费者模型能够有效的运行，而编码中不需要考虑同步和线程间唤醒问题。

阻塞队列的使用场景常常是在生产者和消费者模型中，关于什么是生产者和消费者模型，可以看这篇文章：https://blog.csdn.net/snow_5288/article/details/72794306

10. 死锁

前面我们也提到过死锁，在进行并发编程时，如果使用不当，很容易造成死锁，死锁便会导致这些线程处于等待状态，无法继续执行。当线程进入对象的synchronized代码块时，便占有了资源，直到它退出该代码块或者调用wait方法，才释放资源，在此期间，其他线程将不能进入该代码块。当线程互相持有对方所需要的资源时，会互相等待对方释放资源，如果线程都不主动释放所占有的资源，将产生死锁。当然重入的那种情况除外，在锁的第一篇中有讲解。

10.1 死锁产生的条件

1. 互斥条件：指进程对所分配到的资源进行排它性使用，即在一段时间内某资源只由一个进程占用。如果此时还有其它进程请求资源，则请求者只能等待，直至占有资源的进程用毕释放。

2. 请求和保持条件：一个进程因请求被占用资源而发生阻塞时，对已获得的资源保持不放。

3. 不剥夺条件：任何一个资源在没被该进程释放之前，任何其他进程都无法对他剥夺占用。

4.循环等待条件：当发生死锁时，所等待的进程必定会形成一个环路（类似于死循环），造成永久阻塞。

10.2 死锁的例子

面试时，经常会有面试官问，手写一个死锁代码。其实不用怕，理解后，死锁也就不难了。

private static class SimpleDeadLock {

        private static class A extends Thread {

            private final Object lock1;
            private final Object lock2;

            A(Object lock1, Object lock2) {
                this.lock1 = lock1;
                this.lock2 = lock2;
            }

            @Override
            public void run() {
                // A先拿到了lock1锁
                synchronized (lock1) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    // 休眠1秒后，尝试拿lock2锁，但是lock2锁被B持有无法释放，而B在等待A释放lock1锁才能继续走
                    synchronized (lock2) {
                        System.out.println("Hello A");
                    }
                }
            }
        }

        private static class B extends Thread {

            private final Object lock1;
            private final Object lock2;

            B(Object lock1, Object lock2) {
                this.lock1 = lock1;
                this.lock2 = lock2;
            }

            @Override
            public void run() {
                // B先拿到了lock2锁
                synchronized (lock2) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    // 休眠1秒后，尝试拿lock1锁
                    synchronized (lock1) {
                        System.out.println("Hello B");
                    }
                }
            }
        }

        static void test() {
            final Object lock1 = new Object();
            final Object lock2 = new Object();
            A a = new A(lock1, lock2);
            a.start();
            B b = new B(lock1, lock2);
            b.start();
        }
    }

10.3 如何避免死锁

1. 避免一个线程同时获取多个锁

2. 避免一个线程在锁内同时占用多个资源，尽量保证每个锁只占用一个资源

3. 尝试使用定时锁，使用lock.tryLock来代替使用内置锁。

11. CountdownLatch

CountDownLatch是在java1.5被引入的，跟它一起被引入的并发工具类还有CyclicBarrier、Semaphore(上一章有讲)、ConcurrentHashMap和BlockingQueue，它们都存在于java.util.concurrent包下。CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

11.1 先看一个例子

private static class CountdownLatchTest {

        private final static int THREAD_NUM = 10;

        private static class CountdownLatchTask implements Runnable {

            private final CountDownLatch countDownLatch;
            private final String threadName;

            CountdownLatchTask(CountDownLatch countDownLatch, String threadName) {
                this.countDownLatch = countDownLatch;
                this.threadName = threadName;
            }

            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(threadName + " 执行完毕");
                countDownLatch.countDown();
            }
        }

        static void test() {
            CountDownLatch countDownLatch = new CountDownLatch(THREAD_NUM);
            ExecutorService exec = Executors.newCachedThreadPool();
            for (int i = 0; i < THREAD_NUM; i++) {
                exec.execute(new CountdownLatchTask(countDownLatch, "thread - " + i));
            }
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("大家都执行完成了，做总结性工作");
            exec.shutdown();
        }
    }

执行输出：

thread - 4 执行完毕

thread - 0 执行完毕

thread - 3 执行完毕

thread - 1 执行完毕

thread - 2 执行完毕

thread - 5 执行完毕

thread - 6 执行完毕

thread - 7 执行完毕

thread - 9 执行完毕

thread - 8 执行完毕

大家都执行完成了，做总结性工作

11.2 CountdownLatch源码分析

public class CountDownLatch {
    
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

    private final Sync sync;

    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

    public void countDown() {
        sync.releaseShared(1);
    }

    public long getCount() {
        return sync.getCount();
    }
    public String toString() {
        return super.toString() + "[Count = " + sync.getCount() + "]";
    }
}

当调用了countDownLatch.await()方法后，当前线程就会进入了一个死循环当中，在这个死循环里面，会不断的进行判断，通过调用tryAcquireShared方法，不断判断我们上面说的那个计数器，看看它的值是否为0了(为0的时候，其实就是我们调用了足够多次数的countDownLatch.countDown()方法的时候)，如果是为0的话，tryAcquireShared就会返回1，然后跳出了循环，也就不再"阻塞"当前线程了。需要注意的是，说是在不停的循环，其实也并非在不停的执行for循环里面的内容，因为在后面调用parkAndCheckInterrupt()方法时，在这个方法里面是会调用 LockSupport.park(this) 来禁用当前线程。

11.3 CountdownLatch使用场景

1. 实现最大的并行性：有时我们想同时启动多个线程，实现最大程度的并行性。例如，我们想测试一个单例类。如果我们创建一个初始计数为1的CountDownLatch，并让所有线程都在这个锁上等待，那么我们可以很轻松地完成测试。我们只需调用一次countDown()方法就可以让所有的等待线程同时恢复执行。

2. 开始执行前等待n个线程完成各自任务：例如应用程序启动类要确保在处理用户请求前，所有N个外部系统已经启动和运行了。

3. 死锁检测：一个非常方便的使用场景是，你可以使用n个线程访问共享资源，在每次测试阶段的线程数目是不同的，并尝试产生死锁。

12.CyclicBarrier

CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

12.1 CyclicBarrier例子

private final static int THREAD_NUM = 10;

    private static class CountdownLatchTask implements Runnable {

        private final CyclicBarrier lock;
        private final String threadName;

        CountdownLatchTask(CyclicBarrier lock, String threadName) {
            this.lock = lock;
            this.threadName = threadName;
        }

        @Override
        public void run() {
            System.out.println(threadName + " 准备完成");
            try {
                lock.await();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(threadName + " 执行完成");
        }
    }

    public static void main(String[] args) {
        CyclicBarrier lock = new CyclicBarrier(THREAD_NUM, () -> {
            System.out.println("大家都准备完成了");
        });
        ExecutorService exec = Executors.newCachedThreadPool();
        for (int i = 0; i < THREAD_NUM; i++) {
            exec.submit(new CountdownLatchTask(lock, "Thread-" + i));
        }
        exec.shutdown();
    }

执行输出：

Thread-0 准备完成

Thread-3 准备完成

Thread-2 准备完成

Thread-1 准备完成

Thread-5 准备完成

Thread-4 准备完成

Thread-6 准备完成

Thread-7 准备完成

Thread-8 准备完成

Thread-9 准备完成

大家都准备完成了

Thread-9 执行完成

Thread-0 执行完成

Thread-2 执行完成

Thread-5 执行完成

Thread-8 执行完成

Thread-1 执行完成

Thread-3 执行完成

Thread-7 执行完成

Thread-4 执行完成

Thread-6 执行完成

先让所有线程准备，互相等待，调用cyclicBarrier.await()实现，直到到达某个公共屏障点，然后再一起执行。

12.2 CyclicBarrier使用场景

CyclicBarrier可以用于多线程计算数据，最后合并计算结果的应用场景。比如我们用一个Excel保存了用户所有银行流水，每个Sheet保存一个帐户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完之后，得到每个sheet的日均银行流水，最后，再用barrierAction用这些线程的计算结果，计算出整个Excel的日均银行流水。

12.3 CyclicBarrier与CountdownLatch

看了各种资料和书，大家一致的意见都是CountDownLatch是计数器，只能使用一次，而CyclicBarrier的计数器提供reset功能，可以多次使用。但是我不那么认为它们之间的区别仅仅就是这么简单的一点。我们来从jdk作者设计的目的来看，javadoc是这么描述它们的：

CountDownLatch: A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

CyclicBarrier : A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.

从javadoc的描述可以得出：

CountDownLatch：一个或者多个线程，等待其他多个线程完成某件事情之后才能执行；
CyclicBarrier：多个线程互相等待，直到到达同一个同步点，再继续一起执行。

对于CountDownLatch来说，重点是"一个或者多个线程等待"，而其他的N个线程在完成"某件事情"之后，可以终止，也可以等待。而对于CyclicBarrier，重点是多个线程，在任意一个线程没有完成，所有的线程都必须等待。CountDownLatch是计数器，线程完成一个记录一个，只不过计数不是递增而是递减，而CyclicBarrier更像是一个阀门，需要所有线程都到达，阀门才能打开，然后继续执行。

Java篇 - 锁机制与应用场景全集3 (阻塞队列, 死锁, CountdownLatch, CyclicBarrier)

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method