7.8 数据结构-Java Queue

数据结构-Queue

• 队列
  • BlockingQueue(接口, 阻塞队列)
    • ArrayBlockingQueue
      • Object[] items, 数组存储队列元素
      • add, 添加元素到队列, 添加成功返回true, 如果容量满了则抛IllegalStateException异常, 线程安全

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

      • offer, 添加元素到队列, 添加成功返回true, 否则返回false

        public boolean offer(E e) {
          Objects.requireNonNull(e);
          final ReentrantLock lock = this.lock;
          lock.lock();
          try {
              if (count == items.length)
                  return false;
              else {
                  enqueue(e);
                  return true;
              }
          } finally {
              lock.unlock();
          }
        }
      
        private void enqueue(E x) {
          final Object[] items = this.items;
          items[putIndex] = x;
          if (++putIndex == items.length) putIndex = 0; // 循环
          count++;
          notEmpty.signal(); // take()没有数据被阻塞, 通知take()
        }
      

      • put, 添加元素到队列, 队列满了则阻塞线程, 直到队列数据被消费才被唤醒

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

      • poll, 如果队列为空返回null, 否则返回队头元素

      public E poll() {
          final ReentrantLock lock = this.lock;
          lock.lock();
          try {
              return (count == 0) ? null : dequeue();
          } finally {
              lock.unlock();
          }
      }
      
      private E dequeue() {
        ...
        final Object[] items = this.items;
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length) takeIndex = 0;
        count--;
        ...
        notFull.signal(); // put()队列满时被阻塞, 则唤醒
        return x;
      }
      

      • take, 如果队列为空则阻塞, 否则返回对头元素

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

    • LinkedBlockingQueue
      • Node<E> last, 使用链表存储元素
      • ReentrantLock takeLock, “读”锁, 用于take()、poll()
      • ReentrantLock putLock, “写”锁, 用于put()、offer()
      • AtomicInteger count, 读写时, 获取当前队列的长度
      • add, 添加元素到队列, 添加成功返回true, 如果容量满了则抛IllegalStateException异常, 线程安全
      • put, 添加元素到队列, 队列满了则阻塞线程, 直到队列数据被消费才被唤醒
      • offer, 添加元素到队列, 添加成功返回true, 否则返回false.
      • take, 如果队列为空则阻塞, 否则返回对头元素
      • poll, 如果队列为空返回null, 否则返回队头元素
  • ConcurrentLinkedQueue
  • ProrityBlockingQueue
  • SynchronousQueue

参考

• 系统高效地学习数据结构与算法
• Java阻塞队列ArrayBlockingQueue和LinkedBlockingQueue实现原理分析