LinkedBlockingQueue源码学习-阿里云开发者社区

首先来看一个例子，例子来源于网上：

/*** 多线程模拟实现生产者／消费者模型*  */publicclassBlockingQueueTest2 {
/*** * 定义装苹果的篮子* */publicclassBasket {
// 篮子，能够容纳3个苹果BlockingQueue<String>basket=newLinkedBlockingQueue<String>(3);
// 生产苹果，放入篮子publicvoidproduce() throwsInterruptedException {
// put方法放入一个苹果，若basket满了，等到basket有位置basket.put("An apple");
        }
// 消费苹果，从篮子中取走publicStringconsume() throwsInterruptedException {
// take方法取出一个苹果，若basket为空，等到basket有苹果为止(获取并移除此队列的头部)returnbasket.take();
        }
    }
// 定义苹果生产者classProducerimplementsRunnable {
privateStringinstance;
privateBasketbasket;
publicProducer(Stringinstance, Basketbasket) {
this.instance=instance;
this.basket=basket;
        }
publicvoidrun() {
try {
while (true) {
// 生产苹果System.out.println("生产者准备生产苹果："+instance);
basket.produce();
System.out.println("!生产者生产苹果完毕："+instance);
// 休眠300msThread.sleep(300);
                }
            } catch (InterruptedExceptionex) {
System.out.println("Producer Interrupted");
            }
        }
    }
// 定义苹果消费者classConsumerimplementsRunnable {
privateStringinstance;
privateBasketbasket;
publicConsumer(Stringinstance, Basketbasket) {
this.instance=instance;
this.basket=basket;
        }
publicvoidrun() {
try {
while (true) {
// 消费苹果System.out.println("消费者准备消费苹果："+instance);
System.out.println(basket.consume());
System.out.println("!消费者消费苹果完毕："+instance);
// 休眠1000msThread.sleep(1000);
                }
            } catch (InterruptedExceptionex) {
System.out.println("Consumer Interrupted");
            }
        }
    }
publicstaticvoidmain(String[] args) {
BlockingQueueTest2test=newBlockingQueueTest2();
// 建立一个装苹果的篮子Basketbasket=test.newBasket();
ExecutorServiceservice=Executors.newCachedThreadPool();
Producerproducer=test.newProducer("生产者001", basket);
Producerproducer2=test.newProducer("生产者002", basket);
Consumerconsumer=test.newConsumer("消费者001", basket);
service.submit(producer);
service.submit(producer2);
service.submit(consumer);
// 程序运行5s后，所有任务停止//        try {//            Thread.sleep(1000 * 5);//        } catch (InterruptedException e) {//            e.printStackTrace();//        }//        service.shutdownNow();    }
}

采用线程池和阻塞队列实现生产/消费者模型。其中LinkedBlockingQueue是阻塞队列，同时线程安全，其特点：

采用链表数据结构Node的方式进行节点数据的记录，

同时其进行入队和出队的计数器采用原子性的AtomicInteger

其出队和入队采用采用两把锁，putLock和takeLock，同时进行删除的时候，采用fullLock

其与LinkedBlockingQueue相比，其可以无界可以有界，而ArrayBlockingQueue是有界的，同时实现的数据结构不通过，一个采用数组、一个采用链表，同时采用的锁的方式不同，ArrayBlockingQueue采用一把锁，没有对生产和消费消息进行锁的分离。

1.相关变量

//容量，为空时使用Integer.MAX_VALUE=2^31-1privatefinalintcapacity;
/** Current number of elements *///计数,队列中的元素个数privatefinalAtomicIntegercount=newAtomicInteger();
//头结点，head.item==null，首节点不存放元素transientNode<E>head;
//尾节点，last.next==nullprivatetransientNode<E>last;
/** Lock held by take, poll, etc *///消费队列锁privatefinalReentrantLocktakeLock=newReentrantLock();
/** Wait queue for waiting takes *///消费队列等待消费，用于队满时，进行消费privatefinalConditionnotEmpty=takeLock.newCondition();
/** Lock held by put, offer, etc *///生产队列锁privatefinalReentrantLockputLock=newReentrantLock();
/** Wait queue for waiting puts *///生产队列等待生产，用于队空时，进行生产privatefinalConditionnotFull=putLock.newCondition();
//节点信息：数据、后继点击staticclassNode<E> {
Eitem;
/*** One of:* - the real successor Node* - this Node, meaning the successor is head.next* - null, meaning there is no successor (this is the last node)*///下一个节点，分为三种情况：// 指向真正的节点、指向自己，后继节点为head.next、为空，表示当前节点为尾节点Node<E>next;
Node(Ex) { item=x; }
}

2.构造方法

//构造方法，空参构造默认队列容量为2^31-1publicLinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
//构造方法，带指定容量publicLinkedBlockingQueue(intcapacity) {
//对容量进行校验if (capacity<=0) thrownewIllegalArgumentException();
this.capacity=capacity;
//创建节点信息last=head=newNode<E>(null);
}
//构造方法，放入带指定集合的元素信息入队//首先采用默认大小，进行上锁操作，// 放入元素到队列中，进行遍历，放入publicLinkedBlockingQueue(Collection<?extendsE>c) {
//默认队列大小，2^31-1this(Integer.MAX_VALUE);
//进行上锁操作finalReentrantLockputLock=this.putLock;
putLock.lock(); // Never contended, but necessary for visibilitytry {
//放入元素，进行计数intn=0;
for (Ee : c) {
if (e==null)
thrownewNullPointerException();
if (n==capacity)
thrownewIllegalStateException("Queue full");
enqueue(newNode<E>(e));
++n;
        }
count.set(n);
    } finally {
//释放锁putLock.unlock();
    }
}

3.方法

生产方法

put操作

//入队操作//首先获取锁，再检查队列是否满了，如果满了，则进行阻塞等待，// 如果队列没有满，则进行生产操作，同时计数器进行计数//生产后的元素个数如果还没有达到容量时，会继续唤醒其他生产线程//当生产的元素是元素的第一个元素时唤醒阻塞等待消费的线程publicvoidput(Ee) throwsInterruptedException {
//非空校验if (e==null) thrownewNullPointerException();
// Note: convention in all put/take/etc is to preset local var// holding count negative to indicate failure unless set.//设置计数为0，失败的时候返回intc=-1;
Node<E>node=newNode<E>(e);
finalReentrantLockputLock=this.putLock;
finalAtomicIntegercount=this.count;
//中断上锁putLock.lockInterruptibly();
try {
/** Note that count is used in wait guard even though it is* not protected by lock. This works because count can* only decrease at this point (all other puts are shut* out by lock), and we (or some other waiting put) are* signalled if it ever changes from capacity. Similarly* for all other uses of count in other wait guards.*///检查队列是否满了，满了进行阻塞操作while (count.get() ==capacity) {
notFull.await();
        }
//入队操作，将节点信息插入到队尾//last=last.next=nodeenqueue(node);
c=count.getAndIncrement();
//元素没有满，则唤醒被阻塞的线程，增加线程if (c+1<capacity)
notFull.signal();
    } finally {
//释放锁putLock.unlock();
    }
//插入的是一个元素时唤醒阻塞等待的线程if (c==0)
signalNotEmpty();
}

offer操作

//阻塞带超时时间的offer操作publicbooleanoffer(Ee, longtimeout, TimeUnitunit)
throwsInterruptedException {
if (e==null) thrownewNullPointerException();
longnanos=unit.toNanos(timeout);
intc=-1;
finalReentrantLockputLock=this.putLock;
finalAtomicIntegercount=this.count;
putLock.lockInterruptibly();
try {
while (count.get() ==capacity) {
//如果时间<0，则表示超时返回了，此时队列未满，直接返回if (nanos<=0)
returnfalse;
nanos=notFull.awaitNanos(nanos);
        }
//否者进行入队操作enqueue(newNode<E>(e));
c=count.getAndIncrement();
if (c+1<capacity)
notFull.signal();
    } finally {
putLock.unlock();
    }
if (c==0)
signalNotEmpty();
returntrue;
}
//首先进行非空校验，如果队满了，直接返回false//如果没有满，则进行上锁，同时进行判断，// 如果计数<容量，则进行入队操作//最后释放锁publicbooleanoffer(Ee) {
if (e==null) thrownewNullPointerException();
finalAtomicIntegercount=this.count;
if (count.get() ==capacity)
returnfalse;
intc=-1;
Node<E>node=newNode<E>(e);
finalReentrantLockputLock=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();
returnc>=0;
}

消费者操作

take操作

//take操作 消费消息//如果队列为非空或者被唤醒，进行消费操作，计数器-1publicEtake() throwsInterruptedException {
Ex;
intc=-1;
finalAtomicIntegercount=this.count;
finalReentrantLocktakeLock=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();
returnx;
}

pull操作

//进行消费操作 poll,带超时时间publicEpoll(longtimeout, TimeUnitunit) throwsInterruptedException {
Ex=null;
intc=-1;
longnanos=unit.toNanos(timeout);
finalAtomicIntegercount=this.count;
finalReentrantLocktakeLock=this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() ==0) {
if (nanos<=0)
returnnull;
nanos=notEmpty.awaitNanos(nanos);
        }
x=dequeue();
c=count.getAndDecrement();
if (c>1)
notEmpty.signal();
    } finally {
takeLock.unlock();
    }
if (c==capacity)
signalNotFull();
returnx;
}
//进行poll操作publicEpoll() {
finalAtomicIntegercount=this.count;
if (count.get() ==0)
returnnull;
Ex=null;
intc=-1;
finalReentrantLocktakeLock=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();
returnx;
    }

remove操作

//删除操作，释放指定节点信息publicbooleanremove(Objecto) {
if (o==null) returnfalse;
//对生产消息和消费消息进行上锁fullyLock();
try {
for (Node<E>trail=head, p=trail.next;
p!=null;
trail=p, p=p.next) {
if (o.equals(p.item)) {
//释放节点unlink(p, trail);
returntrue;
            }
        }
returnfalse;
    } finally {
fullyUnlock();
    }
}

drainTo操作

publicintdrainTo(Collection<?superE>c) {
returndrainTo(c, Integer.MAX_VALUE);
    }
//一次性地将队列中的全部元素消费完同时返回指定集合的信息，避免多次加锁造成的性能开销//其中c和maxElement表示返回的集合、要获取的元素个数publicintdrainTo(Collection<?superE>c, intmaxElements) {
if (c==null)
thrownewNullPointerException();
if (c==this)
thrownewIllegalArgumentException();
if (maxElements<=0)
return0;
booleansignalNotFull=false;
//进行上锁finalReentrantLocktakeLock=this.takeLock;
takeLock.lock();
try {
//拿到两者之间的最小的一个intn=Math.min(maxElements, count.get());
// count.get provides visibility to first n NodesNode<E>h=head;
inti=0;
try {
//将元素添加中集合c中while (i<n) {
Node<E>p=h.next;
c.add(p.item);
p.item=null;
h.next=h;
h=p;
++i;
                }
returnn;
            } finally {
// Restore invariants even if c.add() threwif (i>0) {
// assert h.item == null;head=h;
signalNotFull= (count.getAndAdd(-i) ==capacity);
                }
            }
        } finally {
takeLock.unlock();
if (signalNotFull)
signalNotFull();
        }
    }

LinkedBlockingQueue源码学习