Java并发 - Queue

- 日理万妓 2022-09-07 05:49 223阅读 0赞

队列的特点是先进先出，栈的特点是后进先出。Queue继承Collection接口，Stack继承Vector容器类，最顶层接口也是Collection。在Java中容器分为Collection和Map两大类。 Collection家族中除了常见的List、Set，现在又新增一个Queue、Stack。

public interface Queue<E> extends Collection<E> { }

### BlockingQueue 的四组API ###

<table> 
 <thead> 
  <tr> 
   <th>方式</th> 
   <th>抛出异常</th> 
   <th>不抛异常，有返回值</th> 
   <th>阻塞等待</th> 
   <th>超时等待</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>入队列</td> 
   <td>add</td> 
   <td>offer</td> 
   <td>put</td> 
   <td>offer(Element,Time,TimeUnit)</td> 
  </tr> 
  <tr> 
   <td>出队列</td> 
   <td>remove</td> 
   <td>poll</td> 
   <td>take</td> 
   <td>poll(,)</td> 
  </tr> 
  <tr> 
   <td>返回头部元素</td> 
   <td>element</td> 
   <td>peek</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
 </tbody> 
</table>

BlockingQueue主要有两个实现：ArrayBlockingQueue、LinkedBlockingQueue。

BlockingQueue的其中一组核心方法（只介绍其中一组，其它三组类似）：

<table> 
 <thead> 
  <tr> 
   <th>方法名</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>offer(anObject)</td> 
   <td>表示如果可以的话，将anObject加到BlockingQueue里，放入成功返回true，否则返回false。</td> 
  </tr> 
  <tr> 
   <td>offer(E o, long timeout, TimeUnit unit)</td> 
   <td>可以设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。</td> 
  </tr> 
  <tr> 
   <td>put(anObject)</td> 
   <td>把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续。</td> 
  </tr> 
  <tr> 
   <td>poll(time)</td> 
   <td>取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null。</td> 
  </tr> 
  <tr> 
   <td>poll(long timeout, TimeUnit unit)</td> 
   <td>从BlockingQueue取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则直到时间超时还没有数据可取，返回失败。</td> 
  </tr> 
  <tr> 
   <td>take()</td> 
   <td>取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;。</td> 
  </tr> 
  <tr> 
   <td>drainTo()</td> 
   <td>一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数），通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。</td> 
  </tr> 
 </tbody> 
</table>

在ArrayBlockingQueue中对poll(long timeout, TimeUnit unit) 方法的具体实现：

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();
        }
    }

**应用示例**

BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
    // 同步队列不存储元素：一个元素进出完成后，下一个元素才能进出。
    BlockingQueue<String> syncQueue = new SynchronousQueue<>();
    /** * 抛异常: * add()、remove()、element() */
    public void test1() { 
        // ============== add ============= //
        queue.add("1");
        queue.add("2");
        queue.add("3");
        // 再进一个元素则报错
        queue.add("4");
    
        // ============== element ============= //
        // 检查队首元素（该操作不会出队列）
        System.out.println(queue.element());
    
        // ============== remove ============= //
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        // 再取一个
        System.out.println(queue.remove());
    }

/** * 不抛异常，有返回值 * offer()、poll()、peek() */
    public void test2() { 
        // ============== offer ============= //
        queue.offer("1");
        queue.offer("2");
        queue.offer("3");
        // 再添加一个
        System.out.println(queue.offer("4"));
    
        // ============== poll ============= //
        // 检查队首元素（该操作不会出队列）
        System.out.println(queue.peek());
    
        // ============== peek ============= //
        queue.poll();
        queue.poll();
        queue.poll();
        System.out.println(queue.size());
        // 再取一个
        System.out.println(queue.poll());
    }

/** * 阻塞等待 * put、take */
    public void test3() throws InterruptedException { 
        // ============== put ============= //
        queue.put("1");
        queue.put("2");
        queue.put("3");
        // 再添加一个，阻塞
        queue.put("4");
    
        // ============== take ============= //
        queue.take();
        queue.take();
        queue.take();
        System.out.println(queue.size());
        // 再取一个，阻塞
        queue.take();
    }

/** * 超时等待 * offer、poll */
    public void test4() throws InterruptedException { 
        queue.offer("1");
        queue.offer("2");
        queue.offer("3");
        // 2s后入队列失败
        System.out.println(queue.offer("4", 2, TimeUnit.SECONDS));
    
        queue.poll();
        queue.poll();
        queue.poll();
        System.out.println(queue.size());
        // 2s后出队列失败
        System.out.println(queue.poll(2, TimeUnit.SECONDS));
    }

### SynchronousQueue ###

同步队列，不存储元素，一个元素进出完成后，下一个元素才能进出。

/** * 同步队列 * 创建两个线程：线程一存取三个元素，线程二读取三个元素 */
    public void test5() { 
        new Thread(() -> { 
            try { 
                System.out.println(Thread.currentThread().getName() + " put 1");
                syncQueue.put("1");
                System.out.println(Thread.currentThread().getName() + " put 2");
                syncQueue.put("2");
                System.out.println(Thread.currentThread().getName() + " put 3");
                syncQueue.put("3");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }, "A").start();
    	
        new Thread(() -> { 
            try { 
                // 等待两秒再取，确保线程1入队列成功
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName() + "=>" + syncQueue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName() + "=>" + syncQueue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName() + "=>" + syncQueue.take());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }, "B").start();
    }

### PriorityQueue 优先级队列 ###

PriorityQueue是一个带有优先级的队列，而不是先进先出队列，元素按优先级顺序被移除，该队列也没有上限（即 Integer.MAX\_VALUE），无容量限制，自动扩容。

此队列虽然没有容量限制，但是会由于服务器资源耗尽抛OutOfMemoryError异常。

如果队列为空，那么取元素的操作take就会阻塞，所以检索操作take是受阻的。

放入PriorityQueue中的元素需要具有比较能力。

public class PriorityQueueDemo { 
        public static void main(String[] args) { 
            // 设置比对方式
            PriorityQueue<String> priorityQueue = new PriorityQueue<>(new Comparator<String>() { 
                @Override
                public int compare(String o1, String o2) { 
                    return 0;
                }
            });
            priorityQueue.add("c");
            priorityQueue.add("a");
            priorityQueue.add("b");
    
            System.out.println(priorityQueue.poll());
            System.out.println(priorityQueue.poll());
            System.out.println(priorityQueue.poll());
    
    		// 定义MessageObject存放的优先级
            PriorityQueue<MessageObject> MessageObjectQueue = new PriorityQueue<>(new Comparator<MessageObject>() { 
                @Override
                public int compare(MessageObject o1, MessageObject o2) { 
                    // Order比较大的MessageObject放后面
                    return o1.order > o2.order ? -1 : 1;
                }
            });
        }
    
        static class MessageObject { 
            String content;
            int order;
        }
    }

下面探寻一下延时队列的实现原理。

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
        implements BlockingQueue<E> { 
        // 基于PriorityQueue来实现的延时队列
        private final PriorityQueue<E> q = new PriorityQueue<E>();
        ...
    }

DelayQueue的泛型必须实现Delayed接口。

public interface Delayed extends Comparable<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);
    }

**应用示例**

线程池中的定时调度就是使用这样的方法实现的。

public class DelayQueueDemo { 
        public static void main(String[] args) throws InterruptedException { 
            DelayQueue<Message> delayQueue = new DelayQueue<Message>();
            // 这条消息5秒后发送
            Message message = new Message("message - 00001", new Date(System.currentTimeMillis() + 5000L));
            delayQueue.add(message);
    
            while (true) { 
                System.out.println(delayQueue.poll());
                Thread.sleep(1000L);
            }
        }
    
        // 实现Delayed接口的元素才能存到DelayQueue
        static class Message implements Delayed { 
            String content;
            Date sendTime;
    
            /** * @param content 消息内容 * @param sendTime 定时发送 */
            public Message(String content, Date sendTime) { 
                this.content = content;
                this.sendTime = sendTime;
            }
    
            /** * 判断当前这个元素是不是已经到了需要被拿出来的时间 */
            @Override
            public long getDelay(TimeUnit unit) { 
                long duration = sendTime.getTime() - System.currentTimeMillis();
                return TimeUnit.NANOSECONDS.convert(duration, TimeUnit.MILLISECONDS);
            }
    
            @Override
            public int compareTo(Delayed o) { 
                return o.getDelay(TimeUnit.NANOSECONDS) > this.getDelay(TimeUnit.NANOSECONDS) ? 1 : -1;
            }
        }
    }