标签：java并发编程   ati   stat   nod   执行   高性能   err   介绍   协议   

收录于话题
#java 976 #程序员 2286 #并发编程 240 #进阶架构师 | 并发编程专题 12

★★★建议星标我们★★★
公众号改版后文章乱序推荐，希望你可以点击上方“Java进阶架构师”，点击右上角，将我们设为★“星标”！这样才不会错过每日进阶架构文章呀。

2020年Java原创面试题库连载中
【000期】Java最全面试题库思维导图
【001期】JavaSE面试题（一）：面向对象
【002期】JavaSE面试题（二）：基本数据类型与访问修饰符
【003期】JavaSE面试题（三）：JavaSE语法（1）
【004期】JavaSE面试题（四）：JavaSE语法（3）
【005期】JavaSE面试题（五）：String类
【006期】JavaSE面试题（六）：泛型
【007期】JavaSE面试题（七）：异常
【008期】JavaSE面试题（八）：集合之List
【009期】JavaSE面试题（九）：集合之Set
【010期】JavaSE面试题（十）：集合之Map
【011期】JavaSE面试题（十一）：多线程（1）
【012期】JavaSE面试题（十二）：多线程（2）
【013期】JavaSE面试题（十三）：多线程（3）
【014期】JavaSE面试题（十四）：基本IO流
【015期】JavaSE面试题（十五）：网络IO流
【016期】JavaSE面试题（十六）：反射
【017期】JavaSE面试题（十七）：JVM之内存模型
【018期】JavaSE面试题（十八）：JVM之垃圾回收
【020期】JavaSE系列面试题汇总（共18篇）
【019期】JavaWeb面试题（一）：JDBC
【021期】JavaWeb面试题（二）：HTTP协议
【022期】JavaWeb面试题（三）：Cookie和Session
【023期】JavaWeb面试题（四）：JSP
【024期】JavaWeb面试题（五）：Filter和Listener
【025期】Java工具面试题（一）：版本控制工具
【026期】Java工具面试题（二）：项目管理工具
【027期】Java设计模式面试题
【028期】JavaWeb系列面试题汇总（共10篇）
【029期】JavaEE面试题（一）Web应用服务器
【030期】JavaEE面试题（二）SpringMVC
【031期】JavaEE面试题（三）Spring（1）
【032期】JavaEE面试题（四）Spring（2）
【033期】JaveEE面试题（五）MyBatis
【034期】JavaEE面试题（六）Hibernate
【035期】JavaEE面试题（七）SpringBoot（1）
更多内容，点击上面蓝字查看

Java并发编程系列32 | 阻塞队列（下）

阻塞队列在并发编程非常常用，被广泛使用在“生产者-消费者”问题中。本文是阻塞队列下篇。

4.3 SynchronousQueue

SynchronousQueue的同步指的是读线程和写线程需要同步，一个读线程匹配一个写线程。当一个线程往队列中写入一个元素时，写入操作不会立即返回，需要等待另一个线程来将这个元素拿走；当一个读线程做读操作的时候，同样需要一个相匹配的写线程的写操作。
SynchronousQueue 实际不存储元素，数据必须从某个写线程交给某个读线程，而不是写到某个队列中等待被消费。
SynchronousQueue 执行put/take操作时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程），则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。
使用
生产者线程每5秒put一个数据，消费者线程每1秒take一个数据。不管put和take时间如何调整，put和take总是成对出现，SynchronousQueue保证一个读线程匹配一个写线程。

public class SynchronousQueueDemo {
    public static void main(String[] args) {
        BlockingQueue queue = new SynchronousQueue();

        new Thread("生产者") {
            public void run() {
                while (true) {
                    String data = UUID.randomUUID().toString();
                    try {
                        System.out.println("生产者 put: " + data);
                        queue.put(data);
                        Thread.sleep(5000);// 可修改时间测试
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
        }.start();

        new Thread("消费者") {
            public void run() {
                while (true) {
                    try {
                        String data = queue.take();
                        System.out.println("消费者 take: " + data);
                        Thread.sleep(1000);// 可修改时间测试
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            };
        }.start();
    }
}

输出结果：

生产者 put: 890cf163-7c3e-4190-b45e-656cb5757cb5
消费者 take: 890cf163-7c3e-4190-b45e-656cb5757cb5
生产者 put: a858b31c-8bc1-4dce-b5b8-5f1cd318827f
消费者 take: a858b31c-8bc1-4dce-b5b8-5f1cd318827f
生产者 put: 75e6bdd0-a29a-4b70-9d0a-fd2e13fbe479
消费者 take: 75e6bdd0-a29a-4b70-9d0a-fd2e13fbe479
生产者 put: 8db3693e-fe24-4f4e-8f01-eef34542f3ec
消费者 take: 8db3693e-fe24-4f4e-8f01-eef34542f3ec
生产者 put: 233960ce-9ed0-40dd-b450-dd48055191c0
消费者 take: 233960ce-9ed0-40dd-b450-dd48055191c0

类结构：

abstract static class Transferer {
 // 用于转移元素
    abstract Object transfer(Object e, boolean timed, long nanos);
}
// 公平模式
static final class TransferQueue extends Transferer {
 // 等待队列节点
 static final class QNode {
  volatile QNode next; 
  volatile Object item;
  volatile Thread waiter;
  final boolean isData;
 }
}
// 非公平模式
static final class TransferStack extends Transferer {}

put()/take()

public void put(E o) throws InterruptedException {
    if (o == null) throw new NullPointerException();
    if (transferer.transfer(o, false, 0) == null) { // 1
        Thread.interrupted();
        throw new InterruptedException();
    }
}

public E take() throws InterruptedException {
    Object e = transferer.transfer(null, false, 0); // 2
    if (e != null)
        return (E)e;
    Thread.interrupted();
    throw new InterruptedException();
}

可以看到，都是调用Transferer.transfer(E, boolean, long)方法，transfer()就是核心方法了。
transfer()用于转移元素，从生产者手上转到消费者手上，或者消费者调用这个方法来从生产者手上取元素。
第一个参数 e!=null，表示将元素从生产者转移给消费者；如果e==null，表示消费者等待生产者提供元素，然后返回生产者提供的元素。
当调用这个方法时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程，则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。

Object transfer(Object e, boolean timed, long nanos) {
    QNode s = null;
    boolean isData = (e != null);
    for (;;) {
        QNode t = tail;
        QNode h = head;
        if (t == null || h == null)
            continue;

        if (h == t || t.isData == isData) {
            /*
             * 队列为空或队列中节点类型和当前节点一致，节点直接入队
             */
            QNode tn = t.next;
            if (t != tail)// 有其他节点入队
                continue;
            // 有其他节点入队，但是 tail 还是指向原来的，此时设置 tail 即可
            if (tn != null) {
                advanceTail(t, tn);// 如果 tail==t 的话，设置tail=tn
                continue;
            }
            // 
            if (timed && nanos  0)
            --spins;
        // 如果自旋达到了最大的次数，那么检测
        else if (s.waiter == null)
            s.waiter = w;
        // 如果自旋到了最大的次数，那么线程挂起，等待唤醒
        else if (!timed)
            LockSupport.park(this);
        // spinForTimeoutThreshold 这个之前讲 AQS 的时候其实也说过，剩余时间小于这个阈值的时候，就
        // 不要进行挂起了，自旋的性能会比较好
        else if (nanos > spinForTimeoutThreshold)
            LockSupport.parkNanos(this, nanos);
    }
}

4.4 PriorityBlockingQueue

PriorityBlockingQueue队列为***队列，只能指定初始的队列大小，后面插入元素的时候，如果空间不够的话会自动扩容。

PriorityBlockingQueue其实是 PriorityQueue 的线程安全版本，插入队列的对象必须是可比较大小的（comparable）。PriorityBlockingQueue/PriorityQueue 通过堆实现，这里不再详细介绍数据结构，重点讲解阻塞原理。
PriorityQueue 优先级队列的元素按照其自然顺序进行排序或者构造队列时提供的 Comparator 进行排序，插入元素是根据排序规则找到新元素在堆中位置插入。

PriorityBlockingQueue put 方法不会 block，因为它是***队列；take 方法在队列为空的时候会阻塞。
简单看一下put和take方法的阻塞操作，很容易理解。put()：

public void put(E e) {
    offer(e); // never need to block
}

public boolean offer(E e) {
    if (e == null)
        throw new NullPointerException();
    final ReentrantLock lock = this.lock;
    lock.lock();// 获取锁
    int n, cap;
    Object[] array;
    while ((n = size) >= (cap = (array = queue).length))
        tryGrow(array, cap);
    try {
        Comparator super E> cmp = comparator;
        if (cmp == null)
            siftUpComparable(n, e, array);
        else
            siftUpUsingComparator(n, e, array, cmp);
        size = n + 1;
        notEmpty.signal();// 插入元素成功后，唤醒因队列为空而阻塞的读操作线程
    } finally {
        lock.unlock();// 释放锁
    }
    return true;
}

take()：

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();// 获取锁
    E result;
    try {
        /*
         * 队列空时，将当前线程加入notEmpty条件队列阻塞；
         * 当有元素入队时，队列不为空了就可以take出元素，
         * 此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
         */
        while ( (result = dequeue()) == null)
            notEmpty.await();
    } finally {
        lock.unlock();// 释放锁
    }
    return result;
}

4.5 DelayQueue

DelayQueue是一个支持延时获取元素的***阻塞队列。
DelayQueue中的元素都是可延期的，因为必须实现Delayed接口。
插入元素时，会根据延期时间对元素排序，队头的元素是最先到期的；取出元素时，只有在队头元素到期时才能够从队列中取元素。如果队头元素还有t时间到期，则将取出元素线程阻塞t时间，t时间到后再次尝试取出队头元素。
使用
DelayQueue中元素都要实现Delayed接口，getDelay()方法获取延时时间，compareTo()方法比较延时时间用于排序。
将5s、10s、15s后执行的三个item加入DelayQueue队列，从打印结果来看，都是在预期的延时时间从DelayQueue中取出并执行的。

public class DelayQueueTest {
    public static void main(String[] args) throws InterruptedException {
        long curTime = System.currentTimeMillis();
        Item item_5 = new Item("5S后执行的item", curTime + 5000);
        Item item_10 = new Item("10S后执行的item", curTime + 10000);
        Item item_15 = new Item("15S后执行的item", curTime + 15000);
        DelayQueue queue = new DelayQueue();
        queue.put(item_10);
        queue.put(item_15);
        queue.put(item_5);

        System.out.println("开始!!! time=" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
        for (int i = 0; i 

执行结果：

开始!!! time=2019-12-31T12:18:12.361
执行 name=5S后执行的item time=2019-12-31T12:18:17.306
执行 name=10S后执行的item time=2019-12-31T12:18:22.306
执行 name=15S后执行的item time=2019-12-31T12:18:27.306

类结构
DelayQueue使用优先级队列PriorityQueue存储元素；使用ReentrantLock锁，保证队列数据并发环境下的安全性；通过lock的Condition实现阻塞。

public class DelayQueue extends AbstractQueue
  implements BlockingQueue {
 /** 优先级队列，保存元素 */
 private final PriorityQueue q = new PriorityQueue();
 /** 锁，保证队列数据并发环境下的安全性 */
 private final transient ReentrantLock lock = new ReentrantLock();
 /** Condition */
 private final Condition available = lock.newCondition();
 /** 用于优化阻塞 */
 private Thread leader = null;
}

put()：
获取锁lock
添加元素
插入元素为队首元素时，唤醒take线程尝试take元素。因为更新了队首元素，所以要重新检查队首元素是否到期。
释放锁lock

public void put(E e) {
    offer(e);
}

public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();// 获取锁
    try {
        q.offer(e);// 添加元素
        /*
         * 插入元素为队首元素时，唤醒take线程尝试take元素
         * 因为更新了队首元素，所以要重新检查队首元素是否到期
         */
        if (q.peek() == e) {
            leader = null;
            available.signal();
        }
        return true;
    } finally {
        lock.unlock();// 释放锁
    }
}

/**
 * 获取队首元素
 */
public E peek() {
    return (size == 0) ? null : (E) queue[0];
}

take()：
获取锁
如果队列为空，阻塞take线程；插入元素后会take唤醒去获取队首元素。
如果队首元素到期，出队。
如果队首元素未到期，阻塞take线程t时间（t时间就是队首元素的到期剩余时间），时间到后唤醒take线程，尝试获取队首元素出队。
释放锁

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();// 获取锁
    try {
        for (;;) {// 注意是循环
            E first = q.peek();// 获取队首元素
            // 队列为空，take线程阻塞，等待被唤醒后再循环尝试take元素
            if (first == null)
                available.await();
            // 队列不为空
            else {
                long delay = first.getDelay(NANOSECONDS);
                // 队首元素执行时间到了，出队
                if (delay 

5. 总结

阻塞队列是一个比普通队列多出两个附加操作的队列。两个操作分别是：

• 在队列为空时，获取元素的线程会等待队列变为非空。
• 当队列满时，存储元素的线程会等待队列可用。
  ArrayBlockingQueue
  ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
  添加元素时，如果队列满了不能添加元素，就将添加元素的线程阻塞并加入notFull条件队列；当成功删除元素后，队列就可以添加元素了，唤醒notFull条件队列中阻塞的线程，添加元素。
  删除元素时，如果队列空了不能删除元素，就将删除元素的线程阻塞并加入notEmpty条件队列；当成功添加元素后，队列就可以删除元素了，唤醒notEmpty条件队列中阻塞的线程，删除元素。
  LinkedBlockingQueue
  LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE）
• 锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
• 锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。
  ArrayBlockingQueue的读写使用同一个锁来保证数据安全。LinkedBlockingQueue的读写分别用不同的锁来保证数据安全，采用不同的锁可以使读线程和写线程并发执行，提高了吞吐量，但也增加了编程的复杂度。
  SynchronousQueue
  SynchronousQueue的同步指的是读线程和写线程需要同步，一个读线程匹配一个写线程。当一个线程往队列中写入一个元素时，写入操作不会立即返回，需要等待另一个线程来将这个元素拿走；当一个读线程做读操作的时候，同样需要一个相匹配的写线程的写操作。
  SynchronousQueue 实际不存储元素，数据必须从某个写线程交给某个读线程，而不是写到某个队列中等待被消费。
  SynchronousQueue 执行put/take操作时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程），则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。
  PriorityBlockingQueue
  PriorityBlockingQueue队列为***队列，只能指定初始的队列大小，后面插入元素的时候，如果空间不够的话会自动扩容。
  PriorityBlockingQueue其实是 PriorityQueue 的线程安全版本，插入队列的对象必须是可比较大小的（comparable）。PriorityBlockingQueue/PriorityQueue 通过堆实现，这里不再详细介绍数据结构，重点讲解阻塞原理。

  PriorityQueue 优先级队列的元素按照其自然顺序进行排序或者构造队列时提供的 Comparator 进行排序，插入元素是根据排序规则找到新元素在堆中位置插入。

PriorityBlockingQueue put 方法不会 block，因为它是***队列；take 方法在队列为空的时候会阻塞。
DelayQueue
DelayQueue是一个支持延时获取元素的***阻塞队列。
DelayQueue中的元素都是可延期的，因为必须实现Delayed接口。
插入元素时，会根据延期时间对元素排序，队头的元素是最先到期的；取出元素时，只有在队头元素到期时才能够从队列中取元素。如果队头元素还有t时间到期，则将取出元素线程阻塞t时间，t时间到后再次尝试取出队头元素。

并发系列文章汇总

【原创】01|开篇获奖感言
【原创】02|并发编程三大核心问题
【原创】03|重排序-可见性和有序性问题根源
【原创】04|Java 内存模型详解
【原创】05|深入理解 volatile
【原创】06|你不知道的 final
【原创】07|synchronized 原理
【原创】08|synchronized 锁优化
【原创】09|基础干货
【原创】10|线程状态
【原创】11|线程调度
【原创】12|揭秘 CAS
【原创】13|LockSupport
【原创】14|AQS 源码分析
【原创】15|重入锁 ReentrantLock
【原创】16|公平锁与非公平锁
【原创】17|读写锁八讲（上）
【原创】18|读写锁八讲（下）
【原创】19|JDK8新增锁StampedLock
【原创】20|StampedLock源码解析
【原创】21|Condition-Lock的等待通知
【原创】22|倒计时器CountDownLatch
【原创】22|倒计时器CountDownLatch
【原创】23|循环屏障CyclicBarrier
【原创】24|信号量Semaphore
【原创】25|交换器Exchangere
【原创】26|ConcurrentHashMap（上）
【原创】27|ConcurrentHashMap（下）
【原创】28|Copy-On-Write容器
【原创】29|ConcurrentLinkedQueue

之前，给大家发过三份Java面试宝典，这次新增了一份，目前总共是四份面试宝典，相信在跳槽前一个月按照面试宝典准备准备，基本没大问题。
《java面试宝典5.0》(初中级)
《350道Java面试题：整理自100+公司》（中高级）
《资深java面试宝典-视频版》（资深）
《Java[BAT]面试必备》（资深）
分别适用于初中级，中高级，资深级工程师的面试复习。
内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列，高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。

获取方式：点“在看”，V信关注上述Java最全面试题库号并回复 【面试】即可领取，更多精彩陆续奉上。

看到这里，证明有所收获
必须点个在看支持呀，喵

【原创】Java并发编程系列32 | 阻塞队列（下）

标签：java并发编程   ati   stat   nod   执行   高性能   err   介绍   协议   

原文地址：https://blog.51cto.com/15009303/2552575