关于java:BlockingQueue-LinkedBlockingQueue

8次阅读

共计 3552 个字符，预计需要花费 9 分钟才能阅读完成。

LinkedBlockingQueue 是一个能够有界、也能够无界的阻塞队列，以 FIFO（先进先出）的形式拜访队列。

head 是队首节点，是进入队列工夫最长的节点，tail 是队尾节点，是进入队列工夫最短的节点。

节点从队尾退出队列，从 head 出队。

节点： 通过外部类 Node 定义节点：

static class Node<E> {
        E item;
        Node<E> next;

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

很简略，次要属性就两个：
next: 下一节点。
item：节点蕴含的数据。

capacity： 队列容量，无界队列值为 Integer.MAX_VALUE。

count： 队列以后节点数，AtomicInteger 类型。

head： 头节点。

last： 尾结点。

takeLock：ReentrantLock，出队锁，也就是从队列获取数据的锁。

notEmpty：takeLock 的 Condition，帮助获取数据的线程排队。

putLock：ReentrantLock，入队锁，数据退出队列的锁。

notFull：putLock 的 Condition，帮助退出队列的线程排队。

无参构造方法创立一个无边界空队列。

public LinkedBlockingQueue() {this(Integer.MAX_VALUE);
    }

带容量参数 capacity 的构造方法创立一个有界空队列

   public LinkedBlockingQueue(int capacity) {if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(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();
        }
    }

须要留神创立的队列中必然会蕴含一个 dummy 性质的 head 节点，所以出队列的时候这个 head 节点必定也须要跳过。

元素退出队列的 put 办法，代码也比较简单：

 public void put(E e) throws InterruptedException {if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {while (count.get() == capacity) {notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();} finally {putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();}

首先创立 Node，并尝试通过 putLock.lockInterruptibly(); 为以后线程获取对队列的写入锁。如果获取到则会立刻返回，否则以后线程阻塞期待。

拿到写入锁之后，判断如果以后队列已达容量上线（count.get() == capacity）的话，在 notFull 上阻塞期待。

否则，如果容量未满或者其余出队线程（从队列 take 数据的线程胜利拿走数据之后）释放出来队列容量后唤醒以后线程后，以后节点 enqueue(node) 退出队列尾部。而后累加以后队列容量 count，并判断如果容量未满的话，唤醒在 notFull 上期待容量的其余过程。

之后开释 putLock 锁。

最初，如果退出的节点是队列的第一个节点（c==0, 空队列退出），这种状况下可能会有期待 take 数据的线程在阻塞期待，所以调用 signalNotEmpty() 唤醒阻塞的 take 线程。

获取数据的 take 办法：

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 = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();} finally {takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

首先获取 takeLock，获取不到则阻塞期待，获取到则立刻返回。

拿到 take 锁之后，判断如果以后队列是空队列（count.get() == 0）的话，在 notEmpty 上阻塞期待。

否则，如果队列不空或者其余写入线程（参考 put 办法，put 实现之后唤醒 take 阻塞线程）写入队列后唤醒以后线程后，调用 dequeue 办法获取数据。而后从新计算（减）以后队列容量 count，并判断如果队列不空的话，唤醒在 notEmpty 上期待数据的其余 take 过程。

之后开释 takeLock 锁。

最初，如果 take 之前队列已满（c == capacity），这种状况下可能会有期待 put 数据的线程在阻塞期待，signalNotFull() 唤醒阻塞的 put 线程。

最初将通过 dequeue() 办法取出的数据返回。

去 head 的次节点，获取该节点的 item 返回，之后头节点移出队列，次节点变成头节点，并且将变成头节点的次节点（此时其数据 item 曾经被获取到）item 置空。

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

从出队办法 dequeue() 咱们须要明确以下两点：

数据是从次节点（头节点的 next 节点）获取到的，因为头节点是 dummy 节点
头节点移出队列，次节点清理变为 dummy 节点之后，再变为头节点（怯懦的站在头部装 B）

比 SynchronousQueue 的源码简略多了。不过咱们只剖析了他的两个次要办法，对于汇合类的其余办法，LinkedBlockingQueue 也是反对的，代码逻辑也都绝对比较简单，须要留神某些操作比方 remove、contains 等是须要同时获取 putLock 和 takeLock 的。其余的写入办法比方 offer、数据获取办法比方 poll，代码逻辑和 put、take 大同小异，很容易了解。

只不过阻塞队列 LinkedBlockingQueue 也是提供非阻塞办法的，比方 poll、offer，如果 offer 的时候队列已满、或者 poll 的时候空队列，则立刻返回 false。

对代码了解有妨碍的中央应该是 ReentrantLock 和 Condition，是了解 LinkedBlockingQueue 源码的前提。

Thanks a lot!

上一篇 BlockingQueue – 基于 TransferQueue 的 SynchronousQueue

正文完