Stack

基于Vector实现，支持LIFO。

类声明

public class Stack<E> extends Vector<E> {}

push

public E push(E item) {addElement(item);return item;
}

pop

public synchronized E pop() {E  obj;int len = size();obj = peek();removeElementAt(len - 1);return obj;
}

peek

public synchronized E peek() {int len = size();if (len == 0)throw new EmptyStackException();return elementAt(len - 1);
}

Queue

先进先出”（FIFO—first in first out）的线性表

LinkedList类实现了Queue接口，因此我们可以把LinkedList当成Queue来用。

Java里有一个叫做Stack的类，却没有叫做Queue的类（它是个接口名字）。当需要使用栈时，Java已不推荐使用Stack，而是推荐使用更高效的ArrayDeque；既然Queue只是一个接口，当需要使用队列时也就首选ArrayDeque了（次选是LinkedList）。

ArrayDeque和LinkedList是Deque的两个通用实现。

ArrayDeque

（底层是循环数组，有界队列）

head指向首端第一个有效元素，tail指向尾端第一个可以插入元素的空位。因为是循环数组，所以head不一定总等于0，tail也不一定总是比head大。

ConcurrentLinkedQueue

（底层是链表，基于CAS的非阻塞队列，无界队列）

ConcurrentLinkedQueue是一个基于链接节点的无界线程安全队列，它采用先进先出的规则对节点进行排序，当我们添加一个元素的时候，它会添加到队列的尾部，当我们获取一个元素时，它会返回队列头部的元素。它采用了“wait－free”算法（非阻塞）来实现。

1 . 使用 CAS 原子指令来处理对数据的并发访问，这是非阻塞算法得以实现的基础。

2. head/tail 并非总是指向队列的头 / 尾节点，也就是说允许队列处于不一致状态。 这个特性把入队 / 出队时，原本需要一起原子化执行的两个步骤分离开来，从而缩小了入队 / 出队时需要原子化更新值的范围到唯一变量。这是非阻塞算法得以实现的关键。

3. 以批处理方式来更新head/tail，从整体上减少入队 / 出队操作的开销。

4. ConcurrentLinkedQueue的迭代器是弱一致性的，这在并发容器中是比较普遍的现象，主要是指在一个线程在遍历队列结点而另一个线程尝试对某个队列结点进行修改的话不会抛出ConcurrentModificationException，这也就造成在遍历某个尚未被修改的结点时，在next方法返回时可以看到该结点的修改，但在遍历后再对该结点修改时就看不到这种变化。

1. 在入队时最后一个结点中的next域为null

2. 队列中的所有未删除结点的item域不能为null且从head都可以在O(N)时间内遍历到

3. 对于要删除的结点，不是将其引用直接置为空，而是将其的item域先置为null(迭代器在遍历是会跳过item为null的结点)

4. 允许head和tail滞后更新，也就是上文提到的head/tail并非总是指向队列的头 / 尾节点（这主要是为了减少CAS指令执行的次数，但同时会增加volatile读的次数，但是这种消耗较小）。具体而言就是，当在队列中插入一个元素是，会检测tail和最后一个结点之间的距离是否在两个结点及以上(内部称之为hop)；而在出队时，对head的检测就是与队列的第一个结点的距离是否达到两个，有则将head指向第一个结点并将head原来指向的结点的next域指向自己，这样就能断开与队列的联系从而帮助GC

head节点并不是总指向第一个结点，tail也并不是总指向最后一个节点。

 

成员变量

private transient volatile Node<E> head;
private transient volatile Node<E> tail;

 

构造方法

public ConcurrentLinkedQueue() {head = tail = new Node<E>(null);
}

Node#CAS操作

在obj的offset位置比较object field和期望的值，如果相同则更新。这个方法的操作应该是原子的，因此提供了一种不可中断的方式更新object field。

如果node的next值为cmp，则将其更新为val

boolean casNext(Node<E> cmp, Node<E> val) {return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}boolean casItem(E cmp, E val) {return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}private boolean casHead(Node<E> cmp, Node<E> val) {return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
}void lazySetNext(Node<E> val) {UNSAFE.putOrderedObject(this, nextOffset, val);
}

offer（无锁）

/*** Inserts the specified element at the tail of this queue.* As the queue is unbounded, this method will never return {@code false}.** @return {@code true} (as specified by {@link Queue#offer})* @throws NullPointerException if the specified element is null*/
public boolean offer(E e) {checkNotNull(e);final Node<E> newNode = new Node<E>(e);for (Node<E> t = tail, p = t;;) {Node<E> q = p.next;// q/p.next/tail.next为null，则说明p是尾节点，则插入if (q == null) {// CAS插入 p.next = newNode，多线程环境下只有一个线程可以设置成功// 此时 tail.next = newNodeif (p.casNext(null, newNode)) {// CAS成功说明新节点已经放入链表// 如果p不为t，说明当前线程是之前CAS失败后又重试CAS成功的，tail = newNodeif (p != t) // hop two nodes at a timecasTail(t, newNode);  // Failure is OK.return true;}// Lost CAS race to another thread; re-read next}else if (p == q)//多线程操作时候，由于poll时候会把老的head变为自引用，然后head的next变为新head，所以这里需要重新找新的head，因为新的head后面的节点才是激活的节点// p = head , t = tailp = (t != (t = tail)) ? t : head;else// 对上一次CAS失败的线程而言，t.next/p.next/tail.next/q 不是null了// 副作用是p = q，p和q都指向了尾节点，进入第三次循环p = (p != t && t != (t = tail)) ? t : q;}
}

 

poll（无锁）

public E poll() {restartFromHead:for (;;) {for (Node<E> h = head, p = h, q;;) {// 保存当前节点的值E item = p.item;// 当前节点有值则CAS置为null, p.item = nullif (item != null && p.casItem(item, null)) {// CAS成功代表当前节点已经从链表中移除if (p != h) // hop two nodes at a timeupdateHead(h, ((q = p.next) != null) ? q : p);return item;} // 当前队列为空时则返回nullelse if ((q = p.next) == null) {updateHead(h, p);return null;} // 自引用了，则重新找新的队列头节点else if (p == q)continue restartFromHead;elsep = q;}}
}final void updateHead(Node<E> h, Node<E> p) {if (h != p && casHead(h, p))h.lazySetNext(h);
}

peek（无锁）

public E peek() {restartFromHead:for (;;) {for (Node<E> h = head, p = h, q;;) {E item = p.item;if (item != null || (q = p.next) == null) {updateHead(h, p);return item;}else if (p == q)continue restartFromHead;elsep = q;}}
}

size（遍历计算大小，效率低）

public int size() {int count = 0;for (Node<E> p = first(); p != null; p = succ(p))if (p.item != null)// Collection.size() spec says to max outif (++count == Integer.MAX_VALUE)break;return count;
}

PriorityQueue

（底层是数组，逻辑上是小顶堆，无界队列）

PriorityQueue底层实现的数据结构是“堆”，堆具有以下两个性质：

任意一个节点的值总是不大于（最大堆）或者不小于（最小堆）其父节点的值；堆是一棵完全二叉树

基于数组实现的二叉堆，对于数组中任意位置的n上元素，其左孩子在[2n+1]位置上，右孩子[2(n+1)]位置，它的父亲则在[(n-1)/2]上，而根的位置则是[0]。

1）时间复杂度：remove()方法和add()方法时间复杂度为O(logn)，remove(Object obj)和contains()方法需要O(n)时间复杂度，取队头则需要O(1)时间

2）在初始化阶段会执行建堆函数，最终建立的是最小堆，每次出队和入队操作不能保证队列元素的有序性，只能保证队头元素和新插入元素的有序性，如果需要有序输出队列中的元素，则只要调用Arrays.sort()方法即可

3）可以使用Iterator的迭代器方法输出队列中元素

4）PriorityQueue是非同步的，要实现同步需要调用java.util.concurrent包下的PriorityBlockingQueue类来实现同步

5）在队列中不允许使用null元素

6）PriorityQueue默认是一个小顶堆，然而可以通过传入自定义的Comparator函数来实现大顶堆

替代：用TreeMap复杂度太高，有没有更好的方法。hash方法，但是队列不是定长的，如果改变了大小要rehash代价太大，还有什么方法？用堆实现，那每次get put复杂度是多少（lgN）

BlockingQueue

对于许多多线程问题，都可以通过使用一个或多个队列以优雅的方式将其形式化

生产者线程向队列插入元素，消费者线程则取出它们。使用队列，可以安全地从一个线程向另一个线程传递数据。

比如转账

一个线程将转账指令放入队列

一个线程从队列中取出指令执行转账，只有这个线程可以访问银行对象的内部。因此不需要同步

 

当试图向队列中添加元素而队列已满，或是想从队列移出元素而队列为空的时候，阻塞队列导致线程阻塞

在协调多个线程之间的合作时，阻塞队列是很有用的。

工作者线程可以周期性地将中间结果放入阻塞队列，其他工作者线程取出中间结果并进一步修改。队列会自动平衡负载，大概第一个线程集比第二个运行的慢，那么第二个线程集在等待结果时会阻塞，反之亦然

1）LinkedBlockingQueue的容量是没有上边界的，是一个双向队列

2）ArrayBlockingQueue在构造时需要指定容量，并且有一个参数来指定是否需要公平策略

3）PriorityBlockingQueue是一个带优先级的队列，元素按照它们的优先级顺序被移走。该队列没有容量上限。

4）DelayQueue包含实现了Delayed接口的对象

5）TransferQueue接口允许生产者线程等待，直到消费者准备就绪可以接收一个元素。如果生产者调用transfer方法，那么这个调用会阻塞，直到插入的元素被消费者取出之后才停止阻塞。

LinkedTransferQueue类实现了这个接口

ArrayBlockingQueue（底层是数组，阻塞队列，一把锁两个Condition，有界同步队列）

基于数组、先进先出、线程安全的集合类，特点是可实现指定时间的阻塞读写，并且容量是可限制的。

成员变量

/** The queued items */
final Object[] items;/** items index for next take, poll, peek or remove */
int takeIndex;/** items index for next put, offer, or add */
int putIndex;/** Number of elements in the queue */
int count;/** Concurrency control uses the classic two-condition algorithm* found in any textbook.*//** Main lock guarding all access */
final ReentrantLock lock;/** Condition for waiting takes */
private final Condition notEmpty;/** Condition for waiting puts */
private final Condition notFull;/*** Shared state for currently active iterators, or null if there* are known not to be any.  Allows queue operations to update* iterator state.*/
transient Itrs itrs = null;

put（有锁，队列满则阻塞）

public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == items.length)notFull.await();enqueue(e);} finally {lock.unlock();}
}private void enqueue(E x) {// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;final Object[] items = this.items;items[putIndex] = x;if (++putIndex == items.length)putIndex = 0;count++;notEmpty.signal();
}

take（有锁，队列空则阻塞）

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)notEmpty.await();return dequeue();} finally {lock.unlock();}
}private E dequeue() {// assert lock.getHoldCount() == 1;// assert items[takeIndex] != null;final Object[] items = this.items;@SuppressWarnings("unchecked")E x = (E) items[takeIndex];items[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();notFull.signal();return x;
}

offer（有锁，最多阻塞一段时间）

public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {checkNotNull(e);long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == items.length) {if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}enqueue(e);return true;} finally {lock.unlock();}
}

 

poll（有锁，最多阻塞一段时间）

public E poll(long timeout, TimeUnit unit) throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0) {if (nanos <= 0)return null;nanos = notEmpty.awaitNanos(nanos);}return dequeue();} finally {lock.unlock();}
}

peek（有锁）

public E peek() {final ReentrantLock lock = this.lock;lock.lock();try {return itemAt(takeIndex); // null when queue is empty} finally {lock.unlock();}final E itemAt(int i) {return (E) items[i];
}

遍历（构造迭代器加锁，遍历迭代器也加锁）

LinkedBlockingQueue

（底层是链表，阻塞队列，两把锁，各自对应一个Condition，无界同步队列）

另一种BlockingQueue的实现，基于链表，没有容量限制。

由于出队只操作队头，入队只操作队尾，这里巧妙地使用了两把锁，对于put和offer入队操作使用一把锁，对于take和poll出队操作使用一把锁，避免了出队、入队时互相竞争锁的现象，因此LinkedBlockingQueue在高并发读写都多的情况下，性能会较ArrayBlockingQueue好很多，在遍历以及删除的情况下则要两把锁都要锁住。

多CPU情况下可以在同一时刻既消费又生产。

LinkedBlockingDeque

（底层是双向链表，阻塞队列，一把锁两个Condition，无界同步队列）

LinkedBlockingDeque是一个基于链表的双端阻塞队列。和LinkedBlockingQueue类似，区别在于该类实现了Deque接口，而LinkedBlockingQueue实现了Queue接口。

LinkedBlockingDeque内部只有一把锁以及该锁上关联的两个条件，所以可以推断同一时刻只有一个线程可以在队头或者队尾执行入队或出队操作（类似于ArrayBlockingQueue）。可以发现这点和LinkedBlockingQueue不同，LinkedBlockingQueue可以同时有两个线程在两端执行操作。

LinkedBlockingDeque和LinkedBlockingQueue的相同点在于：

1. 基于链表

2. 容量可选，不设置的话，就是Int的最大值

和LinkedBlockingQueue的不同点在于：

1. 双端链表和单链表

2. 不存在哨兵节点

3. 一把锁+两个条件

LinkedBlockingDeque和ArrayBlockingQueue的相同点在于：使用一把锁+两个条件维持队列的同步。

PriorityBlockingQueue

（底层是数组，出队时队空则阻塞；无界队列，不存在队满情况，一把锁一个Condition）

支持优先级的无界阻塞队列。默认情况下元素采用自然顺序升序排序，当然我们也可以通过构造函数来指定Comparator来对元素进行排序。需要注意的是PriorityBlockingQueue不能保证同优先级元素的顺序。

扩容

（基于CAS+Lock，CAS控制创建新的数组原子执行，Lock控制数组替换原子执行）

private void tryGrow(Object[] array, int oldCap) {lock.unlock(); // must release and then re-acquire main lockObject[] newArray = null;if (allocationSpinLock == 0 &&UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,0, 1)) {try {int newCap = oldCap + ((oldCap < 64) ?(oldCap + 2) : // grow faster if small(oldCap >> 1));if (newCap - MAX_ARRAY_SIZE > 0) {    // possible overflowint minCap = oldCap + 1;if (minCap < 0 || minCap > MAX_ARRAY_SIZE)throw new OutOfMemoryError();newCap = MAX_ARRAY_SIZE;}if (newCap > oldCap && queue == array)newArray = new Object[newCap];} finally {allocationSpinLock = 0;}}if (newArray == null) // back off if another thread is allocatingThread.yield();lock.lock();if (newArray != null && queue == array) {queue = newArray;System.arraycopy(array, 0, newArray, 0, oldCap);}
}

 

DelayQueue

（底层是PriorityQueue，无界阻塞队列，过期元素方可移除，基于Lock）

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>implements BlockingQueue<E> {private final transient ReentrantLock lock = new ReentrantLock();private final PriorityQueue<E> q = new PriorityQueue<E>();

DelayQueue队列中每个元素都有个过期时间，并且队列是个优先级队列，当从队列获取元素时候，只有过期元素才会出队列。

每个元素都必须实现Delayed接口

public interface Delayed extends Comparable<Delayed> {/*** Returns the remaining delay associated with this object, in the* given time unit.** @param unit the time unit* @return the remaining delay; zero or negative values indicate* that the delay has already elapsed*/long getDelay(TimeUnit unit);
}

getDelay方法返回对象的残留延迟，负值表示延迟结束

元素只有在延迟用完的时候才能从DelayQueue移出。还必须实现Comparable接口。

一个典型场景是重试机制的实现，比如当调用接口失败后，把当前调用信息放入delay=10s的元素，然后把元素放入队列，那么这个队列就是一个重试队列，一个线程通过take方法获取需要重试的接口，take返回则接口进行重试，失败则再次放入队列，同时也可以在元素加上重试次数。

成员变量

private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();private Thread leader = null;private final Condition available = lock.newCondition();

构造方法

public DelayQueue() {}

put

public void put(E e) {offer(e);}
public boolean offer(E e) {final ReentrantLock lock = this.lock;lock.lock();try {q.offer(e);if (q.peek() == e) {leader = null;// 通知最先等待的线程available.signal();}return true;} finally {lock.unlock();}
}

take

获取并移除队列首元素，如果队列没有过期元素则等待。

第一次调用take时候由于队列空，所以调用（2）把当前线程放入available的条件队列等待，当执行offer并且添加的元素就是队首元素时候就会通知最先等待的线程激活，循环重新获取队首元素，这时候first假如不空，则调用getdelay方法看该元素海剩下多少时间就过期了，如果delay<=0则说明已经过期，则直接出队返回。否则看leader是否为null，不为null则说明是其他线程也在执行take则把该线程放入条件队列，否则是当前线程执行的take方法，则调用(5) await直到剩余过期时间到（这期间该线程会释放锁，所以其他线程可以offer添加元素，也可以take阻塞自己），剩余过期时间到后，该线程会重新竞争得到锁，重新进入循环。

（6）说明当前take返回了元素，如果当前队列还有元素则调用singal激活条件队列里面可能有的等待线程。leader那么为null，那么是第一次调用take获取过期元素的线程，第一次调用的线程调用设置等待时间的await方法等待数据过期，后面调用take的线程则调用await直到signal。

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {for (;;) {
// 1）获取但不移除队首元素E first = q.peek();if (first == null)
// 2）无元素，则阻塞 available.await();else {long delay = first.getDelay(NANOSECONDS);
// 3）有元素，且已经过期，则移除if (delay <= 0)return q.poll();first = null; // don't retain ref while waiting
// 4）if (leader != null)available.await();else {Thread thisThread = Thread.currentThread();
// 5）leader = thisThread;try {
// 继续阻塞延迟的时间available.awaitNanos(delay);} finally {if (leader == thisThread)leader = null;}}}}} finally {if (leader == null && q.peek() != null)available.signal();lock.unlock();}
}

SynchronousQueue

（只存储一个元素，阻塞队列，基于CAS）

实现了BlockingQueue，是一个阻塞队列。

一个只存储一个元素的的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入一直处于阻塞状态，吞吐量高于LinkedBlockingQueue。

SynchronousQueue内部并没有数据缓存空间，你不能调用peek()方法来看队列中是否有数据元素，因为数据元素只有当你试着取走的时候才可能存在，不取走而只想偷窥一下是不行的，当然遍历这个队列的操作也是不允许的。队列头元素是第一个排队要插入数据的线程，而不是要交换的数据。数据是在配对的生产者和消费者线程之间直接传递的，并不会将数据缓冲数据到队列中。可以这样来理解：生产者和消费者互相等待对方，握手，然后一起离开。

// 如果为 true，则等待线程以 FIFO 的顺序竞争访问；否则顺序是未指定的。 // SynchronousQueue<Integer> sc =new SynchronousQueue<>(true);//fair - SynchronousQueue<Integer> sc = new SynchronousQueue<>(); // 默认不指定的话是false，不公平的 TransferQueue
（特殊的BlockingQueue）
生产者会一直阻塞直到所添加到队列的元素被某一个消费者所消费（不仅仅是添加到队列里就完事）当我们不想生产者过度生产消息时，TransferQueue可能非常有用，可避免发生OutOfMemory错误。在这样的设计中，消费者的消费能力将决定生产者产生消息的速度。public interface TransferQueue<E> extends BlockingQueue<E> {/*** 立即转交一个元素给消费者，如果此时队列没有消费者，那就false*/boolean tryTransfer(E e);/*** 转交一个元素给消费者，如果此时队列没有消费者，那就阻塞*/void transfer(E e) throws InterruptedException;/*** 带超时的tryTransfer*/boolean tryTransfer(E e, long timeout, TimeUnit unit)throws InterruptedException;/*** 是否有消费者等待接收数据，瞬时状态，不一定准*/boolean hasWaitingConsumer();/*** 返回还有多少个等待的消费者，跟上面那个一样，都是一种瞬时状态，不一定准*/int getWaitingConsumerCount();}

 

LinkedTransferQueue

（底层是链表，阻塞队列，无界同步队列）

LinkedTransferQueue实现了TransferQueue接口，这个接口继承了BlockingQueue。之前BlockingQueue是队列满时再入队会阻塞，而这个接口实现的功能是队列不满时也可以阻塞，实现一种有阻塞的入队功能。

LinkedTransferQueue实际上是ConcurrentLinkedQueue、SynchronousQueue（公平模式）和LinkedBlockingQueue的超集。而且LinkedTransferQueue更好用，因为它不仅仅综合了这几个类的功能，同时也提供了更高效的实现。

Queue实现类之间的区别

非线程安全的：ArrayDeque、LinkedList、PriorityQueue

线程安全的：ConcurrentLinkedQueue、ConcurrentLinkedDeque、ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue

线程安全的又分为阻塞队列和非阻塞队列，阻塞队列提供了put、take等会阻塞当前线程的方法，比如ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue，也有offer、poll等阻塞一段时间候返回的方法；

非阻塞队列是使用CAS机制保证offer、poll等可以线程安全地入队出队，并且不需要加锁，不会阻塞当前线程，比如ConcurrentLinkedQueue、ConcurrentLinkedDeque。

ArrayBlockingQueue和LinkedBlockingQueue 区别

1. 队列中锁的实现不同

    ArrayBlockingQueue实现的队列中的锁是没有分离的，即生产和消费用的是同一个锁；

    LinkedBlockingQueue实现的队列中的锁是分离的，即生产用的是putLock，消费是takeLock

2. 底层实现不同

前者基于数组，后者基于链表

3. 队列边界不同

    ArrayBlockingQueue实现的队列中必须指定队列的大小，是有界队列

    LinkedBlockingQueue实现的队列中可以不指定队列的大小，但是默认是Integer.MAX_VALUE，是无界队列