JUC之阻塞队列

我就是我 2023-01-22 12:48 11阅读 0赞

# 一：阻塞队列 #

**阻塞队列的用法：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70][]

**阻塞队列的根接口为BlockingQueue，实现这个接口的阻塞队列有很多。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70 1][]

阻塞队列也分为有界和无界的区分。判断有界和无界主要是根据初始化的时候队列的大小是否初始化，如果没有指定队列的大小就是无界队列，否则就是有界队列。

## 1：数组阻塞队列ArrayBlockingQueue ##

ArrayBlockingQueue是一个有界的阻塞队列，其内部实现是将对象放到一个数组里，队列内部是以一个先进先出的顺序进行储存，队列中的头元素在所有元素之中是放入时间最久的那个，而尾元素则是最短的那个。

/**
         * Creates an {@code ArrayBlockingQueue} with the given (fixed)
         * capacity and default access policy.
         *
         * @param capacity the capacity of this queue
         * @throws IllegalArgumentException if {@code capacity < 1}
         */
        public ArrayBlockingQueue(int capacity) {
            this(capacity, false);
        }
    
        /**
         * Creates an {@code ArrayBlockingQueue} with the given (fixed)
         * capacity and the specified access policy.
         *
         * @param capacity the capacity of this queue
         * @param fair if {@code true} then queue accesses for threads blocked
         *        on insertion or removal, are processed in FIFO order;
         *        if {@code false} the access order is unspecified.
         * @throws IllegalArgumentException if {@code capacity < 1}
         */
        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();
        }
    
        /**
         * Creates an {@code ArrayBlockingQueue} with the given (fixed)
         * capacity, the specified access policy and initially containing the
         * elements of the given collection,
         * added in traversal order of the collection's iterator.
         *
         * @param capacity the capacity of this queue
         * @param fair if {@code true} then queue accesses for threads blocked
         *        on insertion or removal, are processed in FIFO order;
         *        if {@code false} the access order is unspecified.
         * @param c the collection of elements to initially contain
         * @throws IllegalArgumentException if {@code capacity} is less than
         *         {@code c.size()}, or less than 1.
         * @throws NullPointerException if the specified collection or any
         *         of its elements are null
         */
        public ArrayBlockingQueue(int capacity, boolean fair,
                                  Collection<? extends E> c) {
            this(capacity, fair);
    
            final ReentrantLock lock = this.lock;
            lock.lock(); // Lock only for visibility, not mutual exclusion
            try {
                int i = 0;
                try {
                    for (E e : c) {
                        checkNotNull(e);
                        items[i++] = e;
                    }
                } catch (ArrayIndexOutOfBoundsException ex) {
                    throw new IllegalArgumentException();
                }
                count = i;
                putIndex = (i == capacity) ? 0 : i;
            } finally {
                lock.unlock();
            }
        }

通过源码可以看到创建ArrayBlockingQueue队列的时候必须指定队列的容量，也就是说这是一个有界队列。

## 2：链表阻塞队列LinkedBlockingQueue ##

LinkedBlockingQueue内部以一个链式结构对其元素进行储存，内部以先进先出的顺序对元素进行储存。队列中的头元素在所有元素之中是放入时间最近的那个，而尾元素则是最短的那个。

/**
         * Creates a {@code LinkedBlockingQueue} with a capacity of
         * {@link Integer#MAX_VALUE}.
         */
        public LinkedBlockingQueue() {
            this(Integer.MAX_VALUE);
        }
    
        /**
         * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity.
         *
         * @param capacity the capacity of this queue
         * @throws IllegalArgumentException if {@code capacity} is not greater
         *         than zero
         */
        public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);
        }
    
        /**
         * Creates a {@code LinkedBlockingQueue} with a capacity of
         * {@link Integer#MAX_VALUE}, initially containing the elements of the
         * given collection,
         * added in traversal order of the collection's iterator.
         *
         * @param c the collection of elements to initially contain
         * @throws NullPointerException if the specified collection or any
         *         of its elements are null
         */
        public LinkedBlockingQueue(Collection<? extends E> c) {
            this(Integer.MAX_VALUE);
            final ReentrantLock putLock = this.putLock;
            putLock.lock(); // Never contended, but necessary for visibility
            try {
                int n = 0;
                for (E e : c) {
                    if (e == null)
                        throw new NullPointerException();
                    if (n == capacity)
                        throw new IllegalStateException("Queue full");
                    enqueue(new Node<E>(e));
                    ++n;
                }
                count.set(n);
            } finally {
                putLock.unlock();
            }
        }

通过源码可以看到LinkedBlockingQueue初始化的时候如果不指定队列的大小，则容量的大小会默认为Integer.MAX\_VALUE，当然我们也可以指定一个队列的大小，本质上这个阻塞队列叫有界队列。

## 3：延时阻塞队列DelayQueue ##

DelayQueue对元素进行持有直到一个特定的延迟到期。注入其中的元素必须实现java.util.concurrent.Delayed接口。

/**
         * Returns the remaining delay associated with this object, in the
         * given time unit.
         *
         * @param unit the time unit
         * @return the remaining delay; zero or negative values indicate
         * that the delay has already elapsed
         */
        long getDelay(TimeUnit unit);

根据返回的时间是0或者负值，延迟将被认为过期，该元素将会在DelayQueue的下一次take被调用的时候被释放掉。

/**
         * Retrieves and removes the head of this queue, waiting if necessary
         * until an element with an expired delay is available on this queue.
         *
         * @return the head of this queue
         * @throws InterruptedException {@inheritDoc}
         */
        public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                for (;;) {
                    E first = q.peek();
                    if (first == null)
                        available.await();
                    else {
                        long delay = first.getDelay(NANOSECONDS);
                        if (delay <= 0)
                            return q.poll();
                        first = null; // don't retain ref while waiting
                        if (leader != null)
                            available.await();
                        else {
                            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();
            }
        }

通过源码分析可以看到延迟队列是一个无界队列。

/**
         * Creates a new {@code DelayQueue} that is initially empty.
         */
        public DelayQueue() {}
    
        /**
         * Creates a {@code DelayQueue} initially containing the elements of the
         * given collection of {@link Delayed} instances.
         *
         * @param c the collection of elements to initially contain
         * @throws NullPointerException if the specified collection or any
         *         of its elements are null
         */
        public DelayQueue(Collection<? extends E> c) {
            this.addAll(c);
        }

在下单之后半个小时之内支付完成的延时操作，可以通过延时队列进行实现。

## 4：优先阻塞队列PriorityBlockingQueue ##

PriorityBlockingQueue是一个无界队列，插入的元素必须实现java.util.Comparator接口，因此该队列中的元素的排序就取决于你自己的Comparable实现。当compare()==0的元素并不强制特定行为。

/**
         * Default array capacity.
         */
        private static final int DEFAULT_INITIAL_CAPACITY = 11;
    
       /**
         * Creates a {@code PriorityBlockingQueue} with the default
         * initial capacity (11) that orders its elements according to
         * their {@linkplain Comparable natural ordering}.
         */
        public PriorityBlockingQueue() {
            this(DEFAULT_INITIAL_CAPACITY, null);
        }
    
        /**
         * Creates a {@code PriorityBlockingQueue} with the specified
         * initial capacity that orders its elements according to their
         * {@linkplain Comparable natural ordering}.
         *
         * @param initialCapacity the initial capacity for this priority queue
         * @throws IllegalArgumentException if {@code initialCapacity} is less
         *         than 1
         */
        public PriorityBlockingQueue(int initialCapacity) {
            this(initialCapacity, null);
        }
    
        /**
         * Creates a {@code PriorityBlockingQueue} with the specified initial
         * capacity that orders its elements according to the specified
         * comparator.
         *
         * @param initialCapacity the initial capacity for this priority queue
         * @param  comparator the comparator that will be used to order this
         *         priority queue.  If {@code null}, the {@linkplain Comparable
         *         natural ordering} of the elements will be used.
         * @throws IllegalArgumentException if {@code initialCapacity} is less
         *         than 1
         */
        public PriorityBlockingQueue(int initialCapacity,
                                     Comparator<? super E> comparator) {
            if (initialCapacity < 1)
                throw new IllegalArgumentException();
            this.lock = new ReentrantLock();
            this.notEmpty = lock.newCondition();
            this.comparator = comparator;
            this.queue = new Object[initialCapacity];
        }

根据源码可以看到如果没有指定队列中的容量大小，则会默认创建一个容量为11的队列。这也不能说明就是一个有界队列，

/**
         * Tries to grow array to accommodate at least one more element
         * (but normally expand by about 50%), giving up (allowing retry)
         * on contention (which we expect to be rare). Call only while
         * holding lock.
         *
         * @param array the heap array
         * @param oldCap the length of the array
         */
        private void tryGrow(Object[] array, int oldCap) {
            lock.unlock(); // must release and then re-acquire main lock
            Object[] newArray = null;
            if (allocationSpinLock == 0 &&
                UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
                                         0, 1)) {
                try {
                    int newCap = oldCap + ((oldCap < 64) ?
                                           (oldCap + 2) : // grow faster if small
                                           (oldCap >> 1));
                    if (newCap - MAX_ARRAY_SIZE > 0) {    // possible overflow
                        int minCap = oldCap + 1;
                        if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
                            throw new OutOfMemoryError();
                        newCap = MAX_ARRAY_SIZE;
                    }
                    if (newCap > oldCap && queue == array)
                        newArray = new Object[newCap];
                } finally {
                    allocationSpinLock = 0;
                }
            }
            if (newArray == null) // back off if another thread is allocating
                Thread.yield();
            lock.lock();
            if (newArray != null && queue == array) {
                queue = newArray;
                System.arraycopy(array, 0, newArray, 0, oldCap);
            }
        }

根据源码的tryGrow()方法，可以看到优先队列中有一个扩容机制，所以优先队列其实是一个无界队列。

## 5：同步阻塞队列SynchronousQueue ##

SynchronousQueue是一个特殊的队列，它的内部只能容纳单个元素，如果该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队列中抽走，同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插入了一条新的元素。

## 6：阻塞双端队列BlockingDeque ##

**阻塞双端队列的用法：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70 2][]

## 7：链表双端阻塞队列LinkedBlockingDeque ##

LinkedBlockingDeque实现了BlockingDeque接口，是一个双端队列，两端都可以抽取数据，和链表阻塞队列一样也是一个有界队列。

/**
         * Creates a {@code LinkedBlockingDeque} with a capacity of
         * {@link Integer#MAX_VALUE}.
         */
        public LinkedBlockingDeque() {
            this(Integer.MAX_VALUE);
        }
    
        /**
         * Creates a {@code LinkedBlockingDeque} with the given (fixed) capacity.
         *
         * @param capacity the capacity of this deque
         * @throws IllegalArgumentException if {@code capacity} is less than 1
         */
        public LinkedBlockingDeque(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
        }
    
        /**
         * Creates a {@code LinkedBlockingDeque} with a capacity of
         * {@link Integer#MAX_VALUE}, initially containing the elements of
         * the given collection, added in traversal order of the
         * collection's iterator.
         *
         * @param c the collection of elements to initially contain
         * @throws NullPointerException if the specified collection or any
         *         of its elements are null
         */
        public LinkedBlockingDeque(Collection<? extends E> c) {
            this(Integer.MAX_VALUE);
            final ReentrantLock lock = this.lock;
            lock.lock(); // Never contended, but necessary for visibility
            try {
                for (E e : c) {
                    if (e == null)
                        throw new NullPointerException();
                    if (!linkLast(new Node<E>(e)))
                        throw new IllegalStateException("Deque full");
                }
            } finally {
                lock.unlock();
            }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70]: /images/20221020/855c5bb215184c7eb4d880dac00f9b67.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70 1]: /images/20221020/b72b8356f9e84c47a282cbd22a4914d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NDY5MDU1_size_16_color_FFFFFF_t_70 2]: /images/20221020/4cfe7b6416384f8db55cb7d2c6dc713c.png