乐趣区

关于java:BlockingQueue-LinkedBlockingQueue

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

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

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

LinkedBlockingQueue# 次要属性

节点: 通过外部类 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 办法

元素退出队列的 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 办法

获取数据的 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() 办法取出的数据返回。

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() 咱们须要明确以下两点:

  1. 数据是从次节点(头节点的 next 节点)获取到的,因为头节点是 dummy 节点
  2. 头节点移出队列,次节点清理变为 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

退出移动版