面试准备 -- 线程池队列LinkedBlockingQueue详解

112 阅读 0 评论 74 点赞

我是靠谱客的博主舒服发带，这篇文章主要介绍面试准备 -- 线程池队列LinkedBlockingQueue详解，现在分享给大家，希望可以做个参考。

前面学了 Executor 框架，线程池参数，以及基本线程池的使用，今天来学学面试问的核心 – 线程池中的队列队列。在聊聊 java 线程池一文中，简单介绍了几种常用的阻塞队列，但都是一笔带过，接下来的文章会着重讲线程池中的队列。
由于线程池中使用的队列有多种，接下来会分多篇文章进行学习。

LinkedBlockingQueue队列：
首先我们先看看下图关系图
从上图我们可以很清楚的看到 LinkedBlockingQueue 类中关系，Collection 接口我想大家都很熟悉，平时我们在开发中最常用的 List，Set 顶层也是该接口。按照 java 的语言风格，基本是不断抽象出公共接口，子接口或抽象类则在原有的基础上提供独有的方法。
而 LinkedBlockingQueue 是 BlockingQueue 的实现类，那我们需要先看看 BlockingQueue 提供了哪些方法。

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
public interface BlockingQueue<E> extends Queue<E> {

    //将对象塞入队列，如果塞入成功返回true, 否则返回false。
    boolean add(E e);

    //将对象塞入到队列中，如果设置成功返回true, 否则返回false
    boolean offer(E e);

    //将元素塞入到队列中，如果队列中已经满了，
    //则该方法会一直阻塞，直到队列中有多余的空间。
    void put(E e) throws InterruptedException;

    //将对象塞入队列并设置时间
    //如果塞入成功返回 true, 否则返回 false.
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;

    //从队列中取对象，如果队列中没有对象，
    //线程会一直阻塞，直到队列中有对象，并且该方法取得了该对象。
    E take() throws InterruptedException;

    //在给定的时间里，从队列中获取对象，
    //时间到了直接调用普通的poll方法，为null则直接返回null。
    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

    //获取队列中剩余长度。
    int remainingCapacity();

    //从队列中移除指定的值。
    boolean remove(Object o);

    //判断队列中包含该对象。
    public boolean contains(Object o);

    //将队列中对象，全部移除，并加到传入集合中。
    int drainTo(Collection<? super E> c);

    //指定最多数量限制将队列中对，全部移除，并家到传入的集合中。
    int drainTo(Collection<? super E> c, int maxElements);
}

看完上面的代码，我们基本知道了 LinkedBlockingQueue 队列有什么方法，简单说就是实现 BlockingQueue 接口的提供的方法。LinkedBlockingQueue 源码中包含很多知识点，比如：显示锁和线程间通信的方式。几天前大佬分析过一篇文章里就讲了线程通信的 Condition 接口，锁的话也稍微讲了些，这里就默认都会的情况下来看。

复制代码

1
2
3
4
5
6
7
8
9
 	 /** 显示锁，在将对象从队列中取出时加的锁 */
    private final ReentrantLock takeLock = new ReentrantLock();
	 /**线程间通信，从队列中取对象，如果队列为空时，就阻塞 */
    private final Condition notEmpty = takeLock.newCondition();
    /** 显示锁，在将对象塞入队列时加的锁 */
    private final ReentrantLock putLock = new ReentrantLock();
    /**线程间通信，塞对象入队列，如果队列满时时，就阻塞*/
    private final Condition notFull = putLock.newCondition();

notEmpty 对象在队列从取出值时，如果队列中没有值了，那线程将会堵塞，等待有值进入队列后唤醒线程取值。
notFull 对象对象，如果塞值进队列时，队列已经满了，那线程将会堵塞，直到队列中值被消费，唤醒线程去塞入值。

LinkedBlockingQueue 中有多种入队方法，下面我们来对比下几种入队方式的优劣势：

入队方法	是否阻塞	适合队列
offer	否	有界队列
put	是	都可以

其实入队方法还要个 add 的，但是它还是调用 offer 的入队方法，这里就不介绍。offer 入队方式比较适合有界队列，offer 在队列满的时候，入队失败会返回 false，而 put 的方法在队列满的时候，会将线程阻塞。

下面我们来看看入队操作的源代码实现：
put 方式：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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 {
        	//判断队列是否已经满，满了就乖乖阻塞	
            while (count.get() == capacity) {
                notFull.await();
            }
            //入队操作
            enqueue(node);
            //自增（线程安全）
            c = count.getAndIncrement();
            //判断队列是否超过边界
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
        	//释放锁
            putLock.unlock();
        }
        //唤醒其他阻塞的线程
        if (c == 0)
            signalNotEmpty();
    }

offer 方式，这个方式有两种不同的方法，这里只介绍一种：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public boolean offer(E e) {
		//对象为空时抛出异常
        if (e == null) throw new NullPointerException();
        //原子量，用于自增
        final AtomicInteger count = this.count;
        //判断队列是否已经满了，满了直接返回 false
        if (count.get() == capacity)
            return false;
        int c = -1;
        Node<E> node = new Node<E>(e);
        //入队锁
        final ReentrantLock putLock = this.putLock;
        //显示锁
        putLock.lock();
        try {
        	//判断队列是否已经满了
            if (count.get() < capacity) {
            //入队
                enqueue(node);
                //自增
                c = count.getAndIncrement();
                //入队后再判断队列是否已满
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
        //解锁
            putLock.unlock();
        }
        //唤醒消费线程
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }

入队方法写完，接下来看看出队的方法：
阻塞出队的方法：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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;
    }

非阻塞出队的方法：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 public E poll() {
        final AtomicInteger count = this.count;
        //队列为空时，返回空
        if (count.get() == 0)
            return null;
        E x = null;
        int c = -1;
        final ReentrantLock takeLock = this.takeLock;
        //加锁
        takeLock.lock();
        try {
        	//判断队列中是否有对象
            if (count.get() > 0) {
            //出队操作
                x = dequeue();
                //自减
                c = count.getAndDecrement();
                //唤醒其他出队操作阻塞的线程
                if (c > 1)
                    notEmpty.signal();
            }
        } finally {
        //解锁
            takeLock.unlock();
        }
        //唤醒其他队列满时阻塞的线程
        if (c == capacity)
            signalNotFull();
        return x;
    }