LinkedBlockingQueue1.8源码

旧城等待， 2022-02-12 10:26 235阅读 0赞

使用

private int i;
        LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(3);
    
        class Producer implements Runnable {
            @Override
            public void run() {
                while (true) {
                    try {
                        queue.put(i++);
                        System.out.println("生产,,,,,,剩余容量：" + queue.remainingCapacity());
                        System.out.println("生产,,,,,,剩余容量：" + queue);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        class Consumer implements Runnable {
            @Override
            public void run() {
                while (true) {
                    try {
                        queue.take();
                        System.out.println("消费,,,,剩余容量：" + queue.remainingCapacity());
                        System.out.println("消费,,,,剩余容量：" + queue);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        public static void main(String[] args) {
            Test linkedBlockingQueueTest = new Test();
            new Thread(linkedBlockingQueueTest.new Producer()).start();
            new Thread(linkedBlockingQueueTest.new Consumer()).start();
        }

生产,剩余容量：2  
生产,剩余容量：\[0\]  
生产,剩余容量：1  
生产,剩余容量：\[0, 1\]  
生产,剩余容量：0  
生产,剩余容量：\[0, 1, 2\]  
消费,剩余容量：1  
生产,剩余容量：0  
消费,剩余容量：\[1, 2, 3\]  
生产,剩余容量：\[1, 2, 3\]  
消费,剩余容量：1  
生产,剩余容量：0  
消费,剩余容量：\[2, 3, 4\]  
生产,剩余容量：\[2, 3, 4\]  
消费,剩余容量：1  
生产,剩余容量：0  
消费,剩余容量：\[3, 4, 5\]  
生产,剩余容量：\[3, 4, 5\]  
消费,剩余容量：1

是一个可变范围的链表，队列的元素是先进先出（FIFO），首元素在队列中待的时间最长，尾元素在队列中待的时间最短。新元素添加在队列的尾部，检索操作获得队列的头部。 有更大的吞吐量比起 ，但是更少的可预见性操作在  
并发应用中。该队列的容量大小是可变的，如果在构造函数中没有指定，默认的容量是 ，链表节点会被动态的创建，除非此操作会导致超出容量。此类的它的迭代器实现了 和 接口的所有方法。  
从源码中可以看到，该类继承了AbstractQueue,实现了BlockingQueue,  
一般情况下，在处理多线程的并发问题时，常常用到此类。

## 构造器 ##

public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);
        }

节点

static class Node<E> {
            E item;
            
            Node<E> next;
    
            Node(E x) { item = x; }
        }

容量大小  
private final int capacity;  
如果没有指定，则为Integer.MAX\_VALUE,支持原子操作

当前元素数量  
private final AtomicInteget count = new AtomicInteget();

链表头节点，其前驱节点为null  
transient Node head;

链表尾节点，其后继为null  
private transient Node last;

针对取和添加操作的两把锁及其上的条件

/\**Lock held by tack,poll,etc*/  
private final ReentrantLook takeLock=new ReentrantLock();  
/\**Wait queue for waiting takes*/  
private final Condition notEmpty=takeLock.newCondition();  
/\**Lock held by put,offer,etc*/  
private final ReentrantLock putLock=new ReentrantLock();  
/\**Wait queue for waiting puts*/  
private final Condition notFull=putLock.newCondition();

## 主要方法 ##

/**添加一个元素到队尾，如果队列已满，则一直处于阻塞状态，直到有可用空间
    */
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        //如果当前线程未被中断，则获取锁
        putLock.lockInterruptibly();
        try {
             /*当容量已满时，等待notfull条件释放锁，陷入等
             待状态有两种方法会激活该线程：
            1.  某个put方法添加元素后，发现队列有空余，就调用  notFull.signal()方法激活阻塞线程；
            2.  Take线程取元素时，发现队列已满，取出元素后，也会调用notFull.signal()方法激活阻塞线程*/
            while (count.get() == capacity) {
                notFull.await();
            }
            //把元素node添加到队尾
            enqueue(node);
            c = count.getAndIncrement();
            //发现队列未满，调用notFull.signal()激活阻塞线程的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 = count.getAndDecrement();
            //再次激活notEmpty
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        //如果容量已满，调用signalNotFull(),put线程处于阻塞状态。
        if (c == capacity)
            signalNotFull();
        return x;
    }

其生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的操作下生产者和消费者可以并行的操作队列中的数据，依次来提高整个队列的并发性能。在这里，我们需要注意的是：如果构造一个LinkedBlockingQueue对象，而没有指定其大小，LinkedBlockingQueue会默认为Integer.MAX\_VALUE，这样的话，如果生产者的速度大于消费者的速度，也许还没有等队列阻塞产生，系统内存就已经被消耗殆尽了。