LinkedBlockingQueue阻塞队列

➢ LinkedBlockingQueue阻塞队列 

LinkedBlockingQueue类图 

LinkedBlockingQueue 中也有两个 Node 分别用来存放首尾节点，并且里面有个初始值为 0 的原子变量 count

用来记录队列元素个数，另外里面有两个ReentrantLock的独占锁，分别用来控制元素入队和出队加锁，其中takeLock

用来控制同时只有一个线程可以从队列获取元素，其他线程必须等待，putLock 控制同时只能有一个线程可以获取锁

去添加元素，其他线程必须等待。另外notEmpty和notFull用来实现入队和出队的同步。 另外由于出入队是两个非

公平独占锁，所以可以同时又一个线程入队和一个线程出队，其实这个是个生产者-消费者模型，

/** 通过take取出进行加锁、取出 */ 

private final ReentrantLock takeLock = new ReentrantLock(); 

/** 等待中的队列等待取出 */

private final Condition notEmpty = takeLock.newCondition(); 

/*通过put放置进行加锁、放置*/

private final ReentrantLock putLock = new ReentrantLock();

/** 等待中的队列等待放置 */

private final Condition notFull = putLock.newCondition(); 

/* 记录集合中的个数（计数器） */ 

private final AtomicInteger count = new AtomicInteger(0); 

队列的容量： 

//队列初始容量，Integer最大值 

public static final int   MAX_VALUE = 0x7fffffff;

public LinkedBlockingQueue() { 

 this(Integer.MAX_VALUE); 

}

 public LinkedBlockingQueue(int capacity) { 

 if (capacity <= 0) throw new IllegalArgumentException(); 

 this.capacity = capacity; 

 //初始化首尾节点 

 last = head = new Node<E>(null); 

}

如图默认队列容量为0x7fffffff;用户也可以自己指定容量。 

LinkedBlockingQueue方法 

ps：下面介绍LinkedBlockingQueue用到很多Lock对象。详细可以查找Lock对象的介绍 

✓  带时间的Offer操作-生产者 

在ArrayBlockingQueue中已经简单介绍了Offer()方法，LinkedBlocking的Offer 方法类似，在此就不过多去

介绍。这次我们从介绍下带时间的Offer方法 

public boolean offer(E e, long timeout, TimeUnit unit)

 throws InterruptedException { 

 //空元素抛空指针异常 

 if (e == null) throw new NullPointerException(); 

 long nanos = unit.toNanos(timeout); 

 int c = -1;

 final ReentrantLock putLock = this.putLock; 

 final AtomicInteger count = this.count;

 //获取可被中断锁，只有一个线程克获取 

 putLock.lockInterruptibly(); 

 try {

 //如果队列满则进入循环 

 while (count.get() == capacity) { 

 //nanos<=0直接返回 

 if (nanos <= 0) 

 return false; 

 //否者调用await进行等待，超时则返回<=0（1） 

 nanos = notFull.awaitNanos(nanos); 

 }

 //await在超时时间内返回则添加元素（2） 

 enqueue(new Node<E>(e)); 

 c = count.getAndIncrement(); 

 //队列不满则激活其他等待入队线程（3） 

 if (c + 1 < capacity) 

 notFull.signal(); 

 } finally {

 //释放锁 

 putLock.unlock();

 }

 //c==0说明队列里面有一个元素，这时候唤醒出队线程（4） 

 if (c == 0)

 signalNotEmpty();

 return true;

}

private void enqueue(Node<E> node) { 

 last = last.next = node; 

}

 private void signalNotEmpty() { 

 final ReentrantLock takeLock = this.takeLock; 

 takeLock.lock();

 try {

 notEmpty.signal(); 

 } finally {

 takeLock.unlock(); 

 }

 }

✓  带时间的poll操作-消费者 

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

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获取。 

public int size() {

 return count.get();

}

✓  peek操作 

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

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，在删除操作时候由于要遍历队列所以加了双重

锁，也就是在删除过程中不允许入队也不允许出队操作。 

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。 

类结构图： 

可知TaskQueue继承了LinkedBlockingQueue并且泛化类型固定了为Runnalbe.重写了offer,poll，take方法。 tomcat 中有个线程池 ThreadPoolExecutor，在 NIOEndPoint 中当 acceptor 线程接受到请求后，会把任务放入队列，然后poller 线程从队列里面获取任务，然后就把任务放入线程池执行。这个ThreadPoolExecutor中的的一个参数就是TaskQueue。 

先看看ThreadPoolExecutor的参数如果是普通LinkedBlockingQueue是怎么样的执行逻辑： 当调用线程池方法  execute() 方法添加一个任务时： 

l  如果当前运行的线程数量小于  corePoolSize，则创建新线程运行该任务 

l  如果当前运行的线程数量大于或等于  corePoolSize，则将这个任务放入阻塞队列。 

l  如果当前队列满了，并且当前运行的线程数量小于  maximumPoolSize，则创建新线程运行该任务； 

l  如果当前队列满了，并且当前运行的线程数量大于或等于  maximumPoolSize，那么线程池将会抛出

RejectedExecutionException异常。 

如果线程执行完了当前任务，那么会去队列里面获取一个任务来执行，如果任务执行完了，并且当前线程数大于

corePoolSize，那么会根据线程空闲时间keepAliveTime回收一些线程保持线程池corePoolSize个线程。 

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

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方法的逻辑： 

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，因为这会开启新线程执行任务，当然前提是当前线程个数没有达到最大值。 

LinkedBlockingQueue安全分析总结 
仔细思考下阻塞队列是如何实现并发安全的维护队列链表的，先分析下简单的情况就是当队列里面有多个元素时候，由于同时只有一个线程（通过独占锁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 条件队列里面等待最久的一个线程。 

LinkedBlockingQueue简单示例 
并发库中的BlockingQueue 是一个比较好玩的类，顾名思义，就是阻塞队列。该类主要提供了两个方法put()和take()，前者将一个对象放到队列中，如果队列已经满了，就等待直到有空闲节点；后者从head取一个对象，如果没有对象，就等待直到有可取的对象。 

下面的例子比较简单，一个读线程，用于将要处理的文件对象添加到阻塞队列中， 另外四个写线程用于取出文件对象，为了模拟写操作耗时长的特点，特让线程睡眠一段随机长度的时间。另外，该Demo也使用到了线程池和原子整型 （AtomicInteger），AtomicInteger可以在并发情况下达到原子化更新，避免使用了synchronized，而且性能非常高。由 于阻塞队列的 put 和 take 操作会阻塞，为了使线程退出，特在队列中添加了一个“标识”，算法中也叫“哨兵”，当发现这个哨兵后，写线程就退出。 

当然线程池也要显式退出了。 

package concurrent;

 import java.io.File;

 import java.io.FileFilter;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.atomic.AtomicInteger;

 public class TestBlockingQueue {

 static long randomTime() { 

 return (long) (Math.random() * 1000); 

 } 

 public static void main(String[] args) { 

 // 能容纳100个文件 
 final BlockingQueue<File> queue = new LinkedBlockingQueue<File>(100);  // 线程池 
 final ExecutorService exec = Executors.newFixedThreadPool(5); 
 final File root = new File("F:\\JavaLib"); 

 // 完成标志 

 final File exitFile = new File("");

 // 读个数 

 final AtomicInteger rc = new AtomicInteger(); 

 // 写个数 

 final AtomicInteger wc = new AtomicInteger(); 

 // 读线程 

 Runnable read = new Runnable() { 

 public void run() {

 scanFile(root);

 scanFile(exitFile);

 } 

 public void scanFile(File file) { 

 if (file.isDirectory()) {

 File[] files = file.listFiles(new FileFilter() {

 public boolean accept(File pathname) {

 return pathname.isDirectory()

 || pathname.getPath().endsWith(".java");

 }

 });

 for (File one : files)

 scanFile(one);

 } else {

 try {

 int index = rc.incrementAndGet();

 System.out.println("Read0: " + index + " "

 + file.getPath());

 queue.put(file);

 } catch (InterruptedException e) {

 }

 }

 } 

 };

 exec.submit(read); 

 // 四个写线程 

 for (int index = 0; index < 4; index++) {

 // write thread 

 final int NO = index;

 Runnable write = new Runnable() {

 String threadName = "Write" + NO;

 public void run() {

 while (true) {

 try {

 Thread.sleep(randomTime());

 int index = wc.incrementAndGet();

 File file = queue.take();

 // 队列已经无对象 

 if (file == exitFile) {

 // 再次添加"标志"，以让其他线程正常退出 

 queue.put(exitFile);

 break;

 }

 System.out.println(threadName + ": " + index + " "

 + file.getPath());

 } catch (InterruptedException e) {

 }

 }

 }

 }; 

 exec.submit(write);

 }

 exec.shutdown(); 

 }

}