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

java进阶架构师

发布于 2020-08-28 08:43:46

3920

发布于 2020-08-28 08:43:46

文章被收录于专栏：Java进阶架构师Java进阶架构师

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<String> queue = new SynchronousQueue<String>();
        
        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<E> extends Transferer<E> {
 // 等待队列节点
 static final class QNode {
  volatile QNode next; 
  volatile Object item;
  volatile Thread waiter;
  final boolean isData;
 }
}
// 非公平模式
static final class TransferStack<E> extends Transferer<E> {}

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)        // can't wait
                return null;
            if (s == null)
                s = new QNode(e, isData);
            // 将当前节点，插入到 tail 的后面
            if (!t.casNext(null, s))        // failed to link in
                continue;

            // 将当前节点设置为新的 tail
            advanceTail(t, s);              // swing tail and wait
            // 自旋阻塞，直到匹配到节点，返回节点
            Object x = awaitFulfill(s, e, timed, nanos);
            // 到这里，说明之前入队的线程被唤醒了
            if (x == s) {                   // wait was cancelled
                clean(t, s);
                return null;
            }

            if (!s.isOffList()) {           // not already unlinked
                advanceHead(t, s);          // unlink if head
                if (x != null)              // and forget fields
                    s.item = s;
                s.waiter = null;
            }
            return (x != null) ? x : e;

        } else {                            // complementary-mode
            /*
             * 如果队列中有等待节点，而且与当前操作可以匹配，则匹配等待队列的队头，出队，返回相应数据。
             */
            QNode m = h.next;               // node to fulfill
            if (t != tail || m == null || h != head)
                continue;                   // inconsistent read

            Object x = m.item;
            if (isData == (x != null) ||    // m already fulfilled
                x == m ||                   // m cancelled
                !m.casItem(x, e)) {         // lost CAS
                advanceHead(h, m);          // dequeue and retry
                continue;
            }

            advanceHead(h, m);              // successfully fulfilled
            LockSupport.unpark(m.waiter);
            return (x != null) ? x : e;
        }
    }
}

void advanceTail(QNode t, QNode nt) {
    if (tail == t)
        UNSAFE.compareAndSwapObject(this, tailOffset, t, nt);
}

/**
 * 自旋阻塞，直到匹配到节点，返回节点
 */
Object awaitFulfill(QNode s, Object e, boolean timed, long nanos) {

    long lastTime = timed ? System.nanoTime() : 0;
    Thread w = Thread.currentThread();
    // 判断需要自旋的次数，
    int spins = ((head.next == s) ?
                 (timed ? maxTimedSpins : maxUntimedSpins) : 0);
    for (;;) {
        // 如果被中断了，那么取消这个节点
        if (w.isInterrupted())
            // 就是将当前节点 s 中的 item 属性设置为 this
            s.tryCancel(e);
        Object x = s.item;
        // 这里是这个方法的唯一的出口
        if (x != e)
            return x;
        // 如果需要，检测是否超时
        if (timed) {
            long now = System.nanoTime();
            nanos -= now - lastTime;
            lastTime = now;
            if (nanos <= 0) {
                s.tryCancel(e);
                continue;
            }
        }
        if (spins > 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<Item> queue = new DelayQueue<Item>();
        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 < 3; i++) {
            Item take = queue.take();
            System.out.println("执行 name=" + take.name + " time=" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
        }
    }
}

class Item implements Delayed {
    String name;
    private long time;

    public Item(String name, long time) {
        this.name = name;
        this.time = time;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return time - System.currentTimeMillis();
    }

    @Override
    public int compareTo(Delayed o) {
        if (!(o instanceof Item)) {
            return -1;
        }
        return (int)(this.time - ((Item)o).time);
    };
}

执行结果：

开始!!! 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<E extends Delayed> extends AbstractQueue<E>
  implements BlockingQueue<E> {
 /** 优先级队列，保存元素 */
 private final PriorityQueue<E> q = new PriorityQueue<E>();
 /** 锁，保证队列数据并发环境下的安全性 */
 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 <= 0)
                    return q.poll();// 出队
                // 到这里，队列不为空，队首元素执行时间还没到，设置leader
                first = null; // don't retain ref while waiting
                // 在此之前，其他线程已经调用过take()设置过leader了
                if (leader != null)
                    available.await();
                else {
                    /*
                     * 设置当前take()线程为leader，并阻塞delay时间
                     * 阻塞唤醒之后，继续循环，尝试take元素
                     */
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();// 释放锁
    }
}

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时间到后再次尝试取出队头元素。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2020-08-17，如有侵权请联系 cloudcommunity@tencent.com 删除

block