java基础之队列

队列.jpg

双端队列Deque

双端队列, 先看下整体结构

image

如图, 主要是addFirst 和 addLast方法, 有很多类实现了这种方法, 双链表结构, 实现Deque的子类如下:

bq.png

如linkedList实现, 参见上文.

BlockingQueue

他是原理是怎么玩的呢? 我们分析下, 先赏心悦目下

image

对于队列而言, 其入列和出列有阻塞和非阻塞之分, 我简单列举下

[图片上传失败...(image-9eac9e-1598688235027)]

针对阻塞和非阻塞的操作简单列表如下:

image

阻塞队列的实现类如下:

[图片上传失败...(image-32712a-1598688235027)]

现在我们找几个实现进行源码分析

java.util.concurrent.ArrayBlockingQueue, 线程安全的

非阻塞分析:

public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false; // 队列满了 就直接返回插入失败
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : dequeue(); //队列为空了直接返回null
        } finally {
            lock.unlock();
        }
    }

阻塞分析:

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

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();  //队列为空, 进行阻塞;由notEmpty.signal()唤醒
            return dequeue();  //返回值
        } finally {
            lock.unlock();
        }
    }

入列和出列公共代码部分:

 入列:
 private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        notEmpty.signal();
    }

出列:
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return x;
    }

LinkedBlockingQueue和上面逻辑很多相似地方, 我现在把入列和出列的代码不同写下:java.util.concurrent.LinkedBlockingQueue

1. 实现方式: 单链表实现

static class Node<E> {
        E item;

        /**
         * One of:
         * - the real successor Node
         * - this Node, meaning the successor is head.next
         * - null, meaning there is no successor (this is the last node)
         */
        Node<E> next;

        Node(E x) { item = x; }
    }

2. 读写锁分离

 final ReentrantLock takeLock = this.takeLock;
 final ReentrantLock putLock = this.putLock;

 //通过原子操作,保证计数
 final AtomicInteger count = this.count;

源码简析

 public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // 预先设置 c 的值为 -1，表示失败
    int c = -1;
    Node<E> node = new Node<E>(e);
    // 获取写锁
    final ReentrantLock putLock = this.putLock;
    // 获取当前队列的大小
    final AtomicInteger count = this.count;
    // 设置可中断锁
    putLock.lockInterruptibly();
    try {
        // 队列满了
        // 当前线程阻塞，等待其他线程的唤醒（其他线程 take 成功后就会唤醒此处线程）
        while (count.get() == capacity) {
            // 无限期等待
             notFull.await();
        }
        // 新增到队列尾部
        enqueue(node);
        // 获取当前的队列数
        c = count.getAndIncrement();
        // 如果队列未满，尝试唤醒一个put的等待线程
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        // 释放锁
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

public E take() throws InterruptedException {
    E x;
    // 默认负数
    int c = -1;
    // 当前链表的个数
    final AtomicInteger count = this.count;
    //获取读锁
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        // 当队列为空时，阻塞，等待其他线程唤醒 
        while (count.get() == 0) {
            notEmpty.await();
          }
        // 从队列的头部拿出一个元素
        x = dequeue();
        //减一操作，C比真实的队列数据大一
        c = count.getAndDecrement();
        // c 大于 0 ，表示队列有值，可以唤醒之前被阻塞的读线程
        if (c > 1)
            notEmpty.signal();
    } finally {
         // 释放锁
        takeLock.unlock();
    }
    // 队列未满，可以唤醒 put 等待线程~
    if (c == capacity)
        signalNotFull();
    return x;
}
参考
https://zhuanlan.zhihu.com/p/143256197

入列和出列公共代码分析

入列:
   private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node;
    }

 出列:
     private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }