接着上回继续讨论：“并发队列：无界阻塞队列 LinkedBlockingQueue 原理探究（1）”

五、 带超时时间的poll操作-消费者

获取并移除队首元素，在指定的时间内去轮询队列看有没有首元素有则返回，否者超时后返回null。

[Java] 纯文本查看 复制代码

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    E x = null;
    int c = -1;
    long nanos = unit.toNanos(timeout);
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;

    //出队线程获取独占锁
    takeLock.lockInterruptibly();
    try {

        //循环直到队列不为空
        while (count.get() == 0) {

            //超时直接返回null
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }

        //出队，计数器减一
        x = dequeue();
        c = count.getAndDecrement();

        //如果出队前队列不为空则发送信号，激活其他阻塞的出队线程
        if (c > 1)
            notEmpty.signal();
    } finally {
        //释放锁
        takeLock.unlock();
    }

    //当前队列容量为最大值-1则激活入队线程。
    if (c == capacity)
        signalNotFull();
    return x;
}

首先获取独占锁，然后进入循环当当前队列有元素才会退出循环，或者超时了，直接返回null。

超时前退出循环后，就从队列移除元素，然后计数器减去一，如果减去1前队列元素大于1则说明当前移除后队列还有元素，那么就发信号激活其他可能阻塞到当前条件信号的线程。

最后如果减去1前队列元素个数=最大值，那么移除一个后会腾出一个空间来，这时候可以激活可能存在的入队阻塞线程。

六、put操作-生产者

与带超时时间的poll类似不同在于put时候如果当前队列满了它会一直等待其他线程调用notFull.signal才会被唤醒。

七、 take操作-消费者

与带超时时间的poll类似不同在于take时候如果当前队列空了它会一直等待其他线程调用notEmpty.signal()才会被唤醒。

八、 size操作

当前队列元素个数，如代码直接使用原子变量count获取。

[Java] 纯文本查看 复制代码

    public int size() {
        return count.get();
    }

九、peek操作

获取但是不移除当前队列的头元素，没有则返回null。

[Java] 纯文本查看 复制代码

public E peek() {
        //队列空，则返回null
        if (count.get() == 0)
            return null;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            Node<E> first = head.next;
            if (first == null)
                return null;
            else
                return first.item;
        } finally {
            takeLock.unlock();
        }
    }

十、 remove操作

删除队列里面的一个元素，有则删除返回true，没有则返回false，在删除操作时候由于要遍历队列所以加了双重锁，也就是在删除过程中不允许入队也不允许出队操作。

[Java] 纯文本查看 复制代码

public boolean remove(Object o) {
    if (o == null) return false;

    //双重加锁
    fullyLock();
    try {

        //遍历队列找则删除返回true
        for (Node<E> trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
            if (o.equals(p.item)) {
                unlink(p, trail);
                return true;
            }
        }
        //找不到返回false
        return false;
    } finally {
        //解锁
        fullyUnlock();
    }
}

void fullyLock() {
    putLock.lock();
    takeLock.lock();
}

void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

void unlink(Node<E> p, Node<E> trail) {

    p.item = null;
    trail.next = p.next;
    if (last == p)
        last = trail;
    //如果当前队列满，删除后，也不忘记最快的唤醒等待的线程
    if (count.getAndDecrement() == capacity)
        notFull.signal();
}

十一、开源框架中使用

tomcat中任务队列TaskQueue。

11.1 类图结构

可知TaskQueue继承了LinkedBlockingQueue并且泛化类型固定了为Runnalbe.重写了offer,poll，take方法。

11.2 TaskQueue

tomcat中有个线程池ThreadPoolExecutor，在NIOEndPoint中当acceptor线程接受到请求后，会把任务放入队列，然后poller 线程从队列里面获取任务，然后就吧任务放入线程池执行。这个ThreadPoolExecutor中的的一个参数就是TaskQueue。

先看看ThreadPoolExecutor的参数如果是普通LinkedBlockingQueue是怎么样的执行逻辑：

当调用线程池方法 execute() 方法添加一个任务时：

• 如果当前运行的线程数量小于 corePoolSize，则创建新线程运行该任务
• 如果当前运行的线程数量大于或等于 corePoolSize，则将这个任务放入阻塞队列。
• 如果当前队列满了，并且当前运行的线程数量小于 maximumPoolSize，则创建新线程运行该任务；
• 如果当前队列满了，并且当前运行的线程数量大于或等于 maximumPoolSize，那么线程池将会抛出RejectedExecutionException异常。
• 如果线程执行完了当前任务，那么会去队列里面获取一个任务来执行，如果任务执行完了，并且当前线程数大于corePoolSize，那么会根据线程空闲时间keepAliveTime回收一些线程保持线程池corePoolSize个线程。
  
  

首先看下线程池中exectue添加任务时候的逻辑：

[Java] 纯文本查看 复制代码

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    //当前工作线程个数小于core个数则开新线程执行（1）
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //放入队列（2）
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }

    //如果队列满了则开新线程，但是个数要不超过最大值，超过则返回false
    //然后执行reject handler（3）
    else if (!addWorker(command, false))
        reject(command);
}

可知当当前工作线程个数为corePoolSize后，如果在来任务会把任务添加到队列，队列满了或者入队失败了则开启新线程。

然后看看TaskQueue中重写的offer方法的逻辑：

[Java] 纯文本查看 复制代码

public boolean offer(Runnable o) {
    // 如果parent为null则直接调用父类方法
    if (parent==null) return super.offer(o);
    //如果当前线程池中线程个数达到最大，则无条件调用父类方法
    if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
    //如果当前提交的任务小于当前线程池线程数，说明线程用不完，没必要重新开线程
    if (parent.getSubmittedCount()<(parent.getPoolSize())) return super.offer(o);
    //如果当前线程池线程个数>core个数但是小于最大个数，则开新线程代替放入队列
    if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
    //到了这里，无条件调用父类
    return super.offer(o);
}

可知parent.getPoolSize()<parent.getMaximumPoolSize()普通队列会把当前任务放入队列，TAskQueue则是返回false，因为这会开启新线程执行任务，当然前提是当前线程个数没有达到最大值。

然后看下Worker线程中如果从队列里面获取任务执行的：

[Java] 纯文本查看 复制代码

final void runWorker(Worker w) {
               ...
        try {
            while (task != null || (task = getTask()) != null) {
               ...
            }
            completedAbruptly = false;
        } finally {
              ...
        }
}

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            ...

            int wc = workerCountOf(c);

            ...

            try {
                //根据timed决定调用poll还是take
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
}

十二、总结

12.1 并发安全总结

仔细思考下阻塞队列是如何实现并发安全的维护队列链表的，先分析下简单的情况就是当队列里面有多个元素时候，由于同时只有一个线程（通过独占锁putLock实现）入队元素并且是操作last节点（，而同时只有一个出队线程(通过独占锁takeLock实现）操作head节点，所以不存在并发安全问题。

• 考虑当队列为空的时候队列状态为：
  
  

这时候假如一个线程调用了take方法,由于队列为空，所以count.get()==0所以当前线程会调用notEmpty.await()把自己挂起，并且放入notEmpty的条件队列，并且释放当前条件变量关联的通过takeLock.lockInterruptibly()获取的独占锁。由于释放了锁，所以这时候其他线程调用take时候就会通过takeLock.lockInterruptibly()获取独占锁，然后同样阻塞到notEmpty.await()，同样会被放入notEmpty的条件队列，也就说在队列为空的情况下可能会有多个线程因为调用take被放入了notEmpty的条件队列。

这时候如果有一个线程调用了put方法，那么就会调用enqueue操作，该操作会在last节点后面添加新元素并且设置last为新节点。然后count.getAndIncrement()先获取当前队列元个数为0保存到c，然后自增count为1，由于c==0所以调用signalNotEmpty激活notEmpty的条件队列里面的阻塞时间最长的线程，这时候take中调用notEmpty.await()的线程会被激活await内部会重新去获取独占锁获取成功则返回，否者被放入AQS的阻塞队列，如果获取成功，那么count.get() >0因为可能多个线程put了，所以调用dequeue从队列获取元素（这时候一定可以获取到），然后调用c = count.getAndDecrement() 把当前计数返回后并减去1，如果c>1 说明当前队列还有其他元素，那么就调用 notEmpty.signal()去激活 notEmpty的条件队列里面的其他阻塞线程。

• 考虑当队列满的时候：
  
  

当队列满的时候调用put方法时候，会由于notFull.await()当前线程被阻塞放入notFull管理的条件队列里面，同理可能会有多个调用put方法的线程都放到了notFull的条件队列里面。

这时候如果有一个线程调用了take方法,调用dequeue()出队一个元素，c = count.getAndDecrement()；count值减一；c==capacity;现在队列有一个空的位置，所以调用signalNotFull()激活notFull条件队列里面等待最久的一个线程。

12.2简单对比

LinkedBlockingQueue与ConcurrentLinkedQueue相比前者前者是阻塞队列使用可重入独占的非公平锁来实现通过使用put锁和take锁使得入队和出队解耦可以同时进行处理，但是同时只有一个线程可以入队或者出队，其他线程必须等待，另外引入了条件变量来进行入队和出队的同步，每个条件变量维护一个条件队列用来存放阻塞的线程，要注意这个队列和AQS的队列不是一个东东。LinkedBlockingQueue的size操作通过使用原子变量count获取能够比较精确的获取当前队列的元素个数，另外remove方法使用双锁保证删除时候队列元素保持不变，另外其实这个是个生产者-消费者模型。

而ConcurrentLinkedQueue则使用CAS非阻塞算法来实现，使用CAS原子操作保证链表构建的安全性，当多个线程并发时候CAS失败的线程不会被阻塞，而是使用cpu资源去轮询CAS直到成功，size方法先比LinkedBlockingQueue的获取的个数是不精确的，因为获取size的时候是通过遍历队列进行的，而遍历过程中可能进行增加删除操作,remove方法操作时候也没有对整个队列加锁，remove时候可能进行增加删除操作，这就可能删除了一个刚刚新增的元素，而不是删除的想要位置的。

接着上回继续讨论：“并发队列：无界阻塞队列 LinkedBlockingQueue 原理探究（1）”

五、 带超时时间的poll操作-消费者

获取并移除队首元素，在指定的时间内去轮询队列看有没有首元素有则返回，否者超时后返回null。

[Java] 纯文本查看 复制代码

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    E x = null;
    int c = -1;
    long nanos = unit.toNanos(timeout);
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;

    //出队线程获取独占锁
    takeLock.lockInterruptibly();
    try {

        //循环直到队列不为空
        while (count.get() == 0) {

            //超时直接返回null
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }

        //出队，计数器减一
        x = dequeue();
        c = count.getAndDecrement();

        //如果出队前队列不为空则发送信号，激活其他阻塞的出队线程
        if (c > 1)
            notEmpty.signal();
    } finally {
        //释放锁
        takeLock.unlock();
    }

    //当前队列容量为最大值-1则激活入队线程。
    if (c == capacity)
        signalNotFull();
    return x;
}

首先获取独占锁，然后进入循环当当前队列有元素才会退出循环，或者超时了，直接返回null。

超时前退出循环后，就从队列移除元素，然后计数器减去一，如果减去1前队列元素大于1则说明当前移除后队列还有元素，那么就发信号激活其他可能阻塞到当前条件信号的线程。

最后如果减去1前队列元素个数=最大值，那么移除一个后会腾出一个空间来，这时候可以激活可能存在的入队阻塞线程。

六、put操作-生产者

与带超时时间的poll类似不同在于put时候如果当前队列满了它会一直等待其他线程调用notFull.signal才会被唤醒。

七、 take操作-消费者

与带超时时间的poll类似不同在于take时候如果当前队列空了它会一直等待其他线程调用notEmpty.signal()才会被唤醒。

八、 size操作

当前队列元素个数，如代码直接使用原子变量count获取。

[Java] 纯文本查看 复制代码

    public int size() {
        return count.get();
    }

九、peek操作

获取但是不移除当前队列的头元素，没有则返回null。

[Java] 纯文本查看 复制代码

public E peek() {
        //队列空，则返回null
        if (count.get() == 0)
            return null;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            Node<E> first = head.next;
            if (first == null)
                return null;
            else
                return first.item;
        } finally {
            takeLock.unlock();
        }
    }

十、 remove操作

删除队列里面的一个元素，有则删除返回true，没有则返回false，在删除操作时候由于要遍历队列所以加了双重锁，也就是在删除过程中不允许入队也不允许出队操作。

[Java] 纯文本查看 复制代码

public boolean remove(Object o) {
    if (o == null) return false;

    //双重加锁
    fullyLock();
    try {

        //遍历队列找则删除返回true
        for (Node<E> trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
            if (o.equals(p.item)) {
                unlink(p, trail);
                return true;
            }
        }
        //找不到返回false
        return false;
    } finally {
        //解锁
        fullyUnlock();
    }
}

void fullyLock() {
    putLock.lock();
    takeLock.lock();
}

void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

void unlink(Node<E> p, Node<E> trail) {

    p.item = null;
    trail.next = p.next;
    if (last == p)
        last = trail;
    //如果当前队列满，删除后，也不忘记最快的唤醒等待的线程
    if (count.getAndDecrement() == capacity)
        notFull.signal();
}

十一、开源框架中使用

tomcat中任务队列TaskQueue。

11.1 类图结构

可知TaskQueue继承了LinkedBlockingQueue并且泛化类型固定了为Runnalbe.重写了offer,poll，take方法。

11.2 TaskQueue

tomcat中有个线程池ThreadPoolExecutor，在NIOEndPoint中当acceptor线程接受到请求后，会把任务放入队列，然后poller 线程从队列里面获取任务，然后就吧任务放入线程池执行。这个ThreadPoolExecutor中的的一个参数就是TaskQueue。

先看看ThreadPoolExecutor的参数如果是普通LinkedBlockingQueue是怎么样的执行逻辑：

当调用线程池方法 execute() 方法添加一个任务时：

• 如果当前运行的线程数量小于 corePoolSize，则创建新线程运行该任务
• 如果当前运行的线程数量大于或等于 corePoolSize，则将这个任务放入阻塞队列。
• 如果当前队列满了，并且当前运行的线程数量小于 maximumPoolSize，则创建新线程运行该任务；
• 如果当前队列满了，并且当前运行的线程数量大于或等于 maximumPoolSize，那么线程池将会抛出RejectedExecutionException异常。
• 如果线程执行完了当前任务，那么会去队列里面获取一个任务来执行，如果任务执行完了，并且当前线程数大于corePoolSize，那么会根据线程空闲时间keepAliveTime回收一些线程保持线程池corePoolSize个线程。
  
  

首先看下线程池中exectue添加任务时候的逻辑：

[Java] 纯文本查看 复制代码

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    //当前工作线程个数小于core个数则开新线程执行（1）
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //放入队列（2）
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }

    //如果队列满了则开新线程，但是个数要不超过最大值，超过则返回false
    //然后执行reject handler（3）
    else if (!addWorker(command, false))
        reject(command);
}

可知当当前工作线程个数为corePoolSize后，如果在来任务会把任务添加到队列，队列满了或者入队失败了则开启新线程。

然后看看TaskQueue中重写的offer方法的逻辑：

[Java] 纯文本查看 复制代码

public boolean offer(Runnable o) {
    // 如果parent为null则直接调用父类方法
    if (parent==null) return super.offer(o);
    //如果当前线程池中线程个数达到最大，则无条件调用父类方法
    if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
    //如果当前提交的任务小于当前线程池线程数，说明线程用不完，没必要重新开线程
    if (parent.getSubmittedCount()<(parent.getPoolSize())) return super.offer(o);
    //如果当前线程池线程个数>core个数但是小于最大个数，则开新线程代替放入队列
    if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
    //到了这里，无条件调用父类
    return super.offer(o);
}

可知parent.getPoolSize()<parent.getMaximumPoolSize()普通队列会把当前任务放入队列，TAskQueue则是返回false，因为这会开启新线程执行任务，当然前提是当前线程个数没有达到最大值。

然后看下Worker线程中如果从队列里面获取任务执行的：

[Java] 纯文本查看 复制代码

final void runWorker(Worker w) {
               ...
        try {
            while (task != null || (task = getTask()) != null) {
               ...
            }
            completedAbruptly = false;
        } finally {
              ...
        }
}

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            ...

            int wc = workerCountOf(c);

            ...

            try {
                //根据timed决定调用poll还是take
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
}

十二、总结

12.1 并发安全总结

仔细思考下阻塞队列是如何实现并发安全的维护队列链表的，先分析下简单的情况就是当队列里面有多个元素时候，由于同时只有一个线程（通过独占锁putLock实现）入队元素并且是操作last节点（，而同时只有一个出队线程(通过独占锁takeLock实现）操作head节点，所以不存在并发安全问题。

• 考虑当队列为空的时候队列状态为：
  
  

这时候假如一个线程调用了take方法,由于队列为空，所以count.get()==0所以当前线程会调用notEmpty.await()把自己挂起，并且放入notEmpty的条件队列，并且释放当前条件变量关联的通过takeLock.lockInterruptibly()获取的独占锁。由于释放了锁，所以这时候其他线程调用take时候就会通过takeLock.lockInterruptibly()获取独占锁，然后同样阻塞到notEmpty.await()，同样会被放入notEmpty的条件队列，也就说在队列为空的情况下可能会有多个线程因为调用take被放入了notEmpty的条件队列。

这时候如果有一个线程调用了put方法，那么就会调用enqueue操作，该操作会在last节点后面添加新元素并且设置last为新节点。然后count.getAndIncrement()先获取当前队列元个数为0保存到c，然后自增count为1，由于c==0所以调用signalNotEmpty激活notEmpty的条件队列里面的阻塞时间最长的线程，这时候take中调用notEmpty.await()的线程会被激活await内部会重新去获取独占锁获取成功则返回，否者被放入AQS的阻塞队列，如果获取成功，那么count.get() >0因为可能多个线程put了，所以调用dequeue从队列获取元素（这时候一定可以获取到），然后调用c = count.getAndDecrement() 把当前计数返回后并减去1，如果c>1 说明当前队列还有其他元素，那么就调用 notEmpty.signal()去激活 notEmpty的条件队列里面的其他阻塞线程。

• 考虑当队列满的时候：
  
  

当队列满的时候调用put方法时候，会由于notFull.await()当前线程被阻塞放入notFull管理的条件队列里面，同理可能会有多个调用put方法的线程都放到了notFull的条件队列里面。

这时候如果有一个线程调用了take方法,调用dequeue()出队一个元素，c = count.getAndDecrement()；count值减一；c==capacity;现在队列有一个空的位置，所以调用signalNotFull()激活notFull条件队列里面等待最久的一个线程。

12.2简单对比

LinkedBlockingQueue与ConcurrentLinkedQueue相比前者前者是阻塞队列使用可重入独占的非公平锁来实现通过使用put锁和take锁使得入队和出队解耦可以同时进行处理，但是同时只有一个线程可以入队或者出队，其他线程必须等待，另外引入了条件变量来进行入队和出队的同步，每个条件变量维护一个条件队列用来存放阻塞的线程，要注意这个队列和AQS的队列不是一个东东。LinkedBlockingQueue的size操作通过使用原子变量count获取能够比较精确的获取当前队列的元素个数，另外remove方法使用双锁保证删除时候队列元素保持不变，另外其实这个是个生产者-消费者模型。

而ConcurrentLinkedQueue则使用CAS非阻塞算法来实现，使用CAS原子操作保证链表构建的安全性，当多个线程并发时候CAS失败的线程不会被阻塞，而是使用cpu资源去轮询CAS直到成功，size方法先比LinkedBlockingQueue的获取的个数是不精确的，因为获取size的时候是通过遍历队列进行的，而遍历过程中可能进行增加删除操作,remove方法操作时候也没有对整个队列加锁，remove时候可能进行增加删除操作，这就可能删除了一个刚刚新增的元素，而不是删除的想要位置的。