Java集合类分析之Queue

public interface Queue<E> extends Collection<E> {  
    boolean add(E e); // 添加元素到队列中，相当于进入队尾排队。 
    boolean offer(E e);  //添加元素到队列中，相当于进入队尾排队.  
    E remove(); //移除队头元素  
    E poll();  //移除队头元素  
    E element(); //获取但不移除队列头的元素  
    E peek();  //获取但不移除队列头的元素  
} 

public interface Deque<E> extends Queue<E> {
boolean add(E e);  
boolean offer(E e);  
void addFirst(E e);  
void addLast(E e);  
boolean offerFirst(E e);  
boolean offerLast(E e);
E removeFirst();   
E removeLast();  
E pollFirst();  
E pollLast();  
E remove();  
E poll();
}

transient Object[] elements;
transient int head;
transient int tail;
private static final int MIN_INITIAL_CAPACITY = 8;
public ArrayDeque() {
        elements = new Object[16];
}
public ArrayDeque(int numElements) {
        allocateElements(numElements);
}
public ArrayDeque(Collection<? extends E> c) {
        allocateElements(c.size());
        addAll(c);
}

private void allocateElements(int numElements) {
        int initialCapacity = MIN_INITIAL_CAPACITY;
        // Find the best power of two to hold elements.
        // Tests "<=" because arrays aren't kept full.
        if (numElements >= initialCapacity) {
            initialCapacity = numElements;
            initialCapacity |= (initialCapacity >>>  1);
            initialCapacity |= (initialCapacity >>>  2);
            initialCapacity |= (initialCapacity >>>  4);
            initialCapacity |= (initialCapacity >>>  8);
            initialCapacity |= (initialCapacity >>> 16);
            initialCapacity++;

            if (initialCapacity < 0)   // Too many elements, must back off
                initialCapacity >>>= 1;// Good luck allocating 2 ^ 30 elements
        }
        elements = new Object[initialCapacity];
}

private void doubleCapacity() {
        assert head == tail;
        int p = head;
        int n = elements.length;
        int r = n - p; // number of elements to the right of p
        int newCapacity = n << 1;
        if (newCapacity < 0)
            throw new IllegalStateException("Sorry, deque too big");
        Object[] a = new Object[newCapacity];
        System.arraycopy(elements, p, a, 0, r);
        System.arraycopy(elements, 0, a, r, p);
        elements = a;
        head = 0;
        tail = n;
}

public boolean add(E e) {  
    addLast(e);  
    return true;  
}  
  
public void addLast(E e) {  
    if (e == null)  
        throw new NullPointerException();  
    elements[tail] = e;  
    if ( (tail = (tail + 1) & (elements.length - 1)) == head)  
        doubleCapacity();  
}  
  
public boolean offer(E e) {  
    return offerLast(e);  
}  
  
public boolean offerLast(E e) {  
    addLast(e);  
    return true;  
}  

public E element() {  
    return getFirst();  
}  
  
public E getFirst() {  
    E x = elements[head];  
    if (x == null)  
        throw new NoSuchElementException();  
    return x;  
}  
  
public E peek() {  
    return peekFirst();  
}  
  
public E peekFirst() {  
    return elements[head]; // elements[head] is null if deque empty  
}  
  
public E getLast() {  
    E x = elements[(tail - 1) & (elements.length - 1)];  
    if (x == null)  
        throw new NoSuchElementException();  
    return x;  
}  
  
public E peekLast() {  
    return elements[(tail - 1) & (elements.length - 1)];  
}

public E getFirst() {  
    final Node<E> f = first;  
    if (f == null)  
        throw new NoSuchElementException();  
    return f.item;  
}  
  
public E getLast() {  
    final Node<E> l = last;  
    if (l == null)  
        throw new NoSuchElementException();  
    return l.item;  
}

public boolean add(E e) {  
    linkLast(e);  
    return true;  
}  
  
public boolean offer(E e) {  
    return add(e);  
}  
  
void linkLast(E e) {  
    final Node<E> l = last;  
    final Node<E> newNode = new Node<>(l, e, null);  
    last = newNode;  
    if (l == null)  
        first = newNode;  
    else  
        l.next = newNode;  
    size++;  
    modCount++;  
} 

transient Object[] queue;
private static final int DEFAULT_INITIAL_CAPACITY = 11;
private void initFromPriorityQueue(PriorityQueue<? extends E> c) {
        if (c.getClass() == PriorityQueue.class) {
            this.queue = c.toArray();
            this.size = c.size();
        } else {
            initFromCollection(c);
        }
}

private void initFromCollection(Collection<? extends E> c) {
        initElementsFromCollection(c);
        heapify();
}
private void initElementsFromCollection(Collection<? extends E> c) {
        Object[] a = c.toArray();
        // If c.toArray incorrectly doesn't return Object[], copy it.
        if (a.getClass() != Object[].class)
            a = Arrays.copyOf(a, a.length, Object[].class);
        int len = a.length;
        if (len == 1 || this.comparator != null)
            for (int i = 0; i < len; i++)
                if (a[i] == null)
                    throw new NullPointerException();
        this.queue = a;
        this.size = a.length;
}
//从插入最后一个元素的父节点位置开始建堆
private void heapify() {
    for (int i = (size >>> 1) - 1; i >= 0; i--)
        siftDown(i, (E) queue[i]);
}
//在位置k插入元素x，为了保持最小堆的性质会不断调整节点位置
private void siftDown(int k, E x) {
    if (comparator != null)
        //使用插入元素的实现的比较器调整节点位置
        siftDownUsingComparator(k, x);
    else
        //使用默认的比较器（按照自然排序规则）调整节点的位置
        siftDownComparable(k, x);
}
//具体实现调整节点位置的函数
private void siftDownComparable(int k, E x) {
    Comparable<? super E> key = (Comparable<? super E>)x;
    // 计算非叶子节点元素的最大位置
    int half = size >>> 1;        // loop while a non-leaf
    //如果不是叶子节点则一直循环
    while (k < half) {
        //得到k位置节点左孩子节点，假设左孩子比右孩子更小
        int child = (k << 1) + 1; // assume left child is least
        //保存左孩子节点值
        Object c = queue[child];
        //右孩子节点的位置
        int right = child + 1;
        //把左右孩子中的较小值保存在变量c中
        if (right < size &&
            ((Comparable<? super E>) c).compareTo((E) queue[right]) > 0)
            c = queue[child = right];
        //如果要插入的节点值比其父节点更小，则交换两个节点的值
        if (key.compareTo((E) c) <= 0)
            break;
        queue[k] = c;
        k = child;
    }
    //循环结束，k是叶子节点
    queue[k] = key;
}

public boolean add(E e) {
    //调用offer函数
    return offer(e);
}
//siftUp之前的代码主要确认队列的容量不发生溢出，并保存队列中的元素个数以及发生结构//性修改的次数
public boolean offer(E e) {
    if (e == null)
        throw new NullPointerException();
    modCount++;
    int i = size;
    if (i >= queue.length)
        grow(i + 1);
    size = i + 1;
    if (i == 0)
        queue[0] = e;
    else
        //具体执行添加元素的函数
        siftUp(i, e);
    return true;
}
//调用不同的比较器调整元素的位置
private void siftUp(int k, E x) {
    if (comparator != null)
        siftUpUsingComparator(k, x);
    else
        siftUpComparable(k, x);
}
//使用默认的比较器调整元素的位置
private void siftUpComparable(int k, E x) {
    Comparable<? super E> key = (Comparable<? super E>) x;
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        //保存父节点的值
        Object e = queue[parent];
        //使用compareTo方法，如果要插入的元素小于父节点的位置则交换两个节点的位置
        if (key.compareTo((E) e) >= 0)
            break;
        queue[k] = e;
        k = parent;
    }
    queue[k] = key;
}
//调用实现的比较器进行元素位置的调整，总的过程和上面一致，就是比较的方法不同
private void siftUpUsingComparator(int k, E x) {
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        Object e = queue[parent];
        //这里是compare方法
        if (comparator.compare(x, (E) e) >= 0)
            break;
        queue[k] = e;
        k = parent;
    }
    queue[k] = x;
}

private E removeAt(int i) {
    assert i >= 0 && i < size;
    modCount++;
    //s是队列的队头，对应到数组中就是最后一个元素
    int s = --size;
    //如果要移除的位置是最后一个位置，则把最后一个元素设为null
    if (s == i) // removed last element
        queue[i] = null;
    else {
        //保存待删除的节点元素
        E moved = (E) queue[s];
        queue[s] = null;
        //先把最后一个元素和i位置的元素交换，之后执行下调方法
        siftDown(i, moved);
        //如果执行下调方法后位置没变，说明该元素是该子树的最小元素，需要执行上调方//法,保持最小堆的性质
        if (queue[i] == moved) {//位置没变
            siftUp(i, moved);   //执行上调方法
            if (queue[i] != moved)//如果上调后i位置发生改变则返回该元素
                return moved;
        }
    }
    return null;
}

/** 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;

    /** Main lock guarding all access */
    final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    /** Condition for waiting puts */
    private final Condition notFull;
public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
}

public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
}
public E remove() {
        E x = poll();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
}
public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
}
public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
}

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(); //入队，唤醒一个空等待线程，类似生产者消费者
}
public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await(); //空了等待
            return dequeue();
        } finally {
            lock.unlock();
        }
}
// dequeue省略，类似enqueue

static final class Node<E> {
        E item;
        Node<E> prev;
        Node<E> next;
        Node(E x) {
            item = x;
        }
}

/** The capacity bound, or Integer.MAX_VALUE if none */
    private final int capacity;
    /** Current number of elements */
    private final AtomicInteger count = new AtomicInteger();
    //两个变量来标示头和尾
    transient Node<E> head;
    private transient Node<E> last;
    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();
    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();
    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

// 插入，不扩容，否则失败
public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;//入队当然用putLock  
        putLock.lock();
        try {
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();//队长度+1 
                if (c + 1 < capacity)
                    notFull.signal();//队列没满，当然可以解锁了
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();//这个方法里发出了notEmpty.signal(); 
        return c >= 0;
}
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 
        takeLock.lock();
        try {
            if (count.get() > 0) {
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();//队列没空，解锁
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();//这个方法里发出了notFull.signal(); 
        return x;
}

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();
}
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;
}

public class ModelSample {
    /** 线程池提交的任务数*/
    private final int taskNum = Runtime.getRuntime().availableProcessors() + 1;
    /** 用于多线程间存取产品的队列*/
    private final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<String>(16);
    /** 记录产量*/
    private final AtomicLong output = new AtomicLong(0);
    /** 记录销量*/
    private final AtomicLong sales = new AtomicLong(0);
    /** 简单的线程起步开关*/
    private final CountDownLatch latch = new CountDownLatch(1);
    /** 停产后是否售完队列内的产品的选项*/
    private final boolean clear;
    /** 用于提交任务的线程池*/
    private final ExecutorService pool;
    /** 简陋的命令发送器*/
    private Scanner scanner;

    public ModelSample(boolean clear) {
        this.pool = Executors.newCachedThreadPool();
        this.clear = clear;
    }

    /**
     * 提交生产和消费任务给线程池,并在准备完毕后等待终止命令
     */
    public void service() {
        doService();
        waitCommand();
    }
    /**
     * 提交生产和消费任务给线程池,并在准备完毕后同时执行
     */
    private void doService() {
        for (int i = 0; i < taskNum; i++) {
            if (i == 0) {
                pool.submit(new Worker(queue, output, latch));
            }
            else {
                pool.submit(new Seller(queue, sales, latch, clear));
            }
        }
        latch.countDown();//开闸放狗,线程池内的线程正式开始工作
    }

    /**
     * 接收来自终端输入的终止命令
     */
    private void waitCommand() {
        scanner = new Scanner(System.in);
        while (!scanner.nextLine().equals("q")) {
            try {
                Thread.sleep(500);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        scanner.close();
        destory();
    }
    /**
     * 停止一切生产和销售的线程
     */
    private void destory() {
        pool.shutdownNow(); //不再接受新任务,同时试图中断池内正在执行的任务
        while (clear && queue.size() > 0) {
            try {
                Thread.sleep(500);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("Products:" + output.get() + "; Sales:" + sales.get());
    }

    public static void main(String[] args) {
        ModelSample model = new ModelSample(false);
        model.service();
    }
}
/**
 * 生产者
 */
class Worker implements Runnable {

    /** 假想的产品*/
    private static final String PRODUCT = "Thinkpad";
    private final LinkedBlockingQueue<String> queue;
    private final CountDownLatch latch;
    private final AtomicLong output;

    public Worker(LinkedBlockingQueue<String> queue, AtomicLong output,CountDownLatch latch) {
        this.output = output;
        this.queue = queue;
        this.latch = latch;
    }

    public void run() {
        try {
            latch.await(); // 放闸之前老实的等待着
            for (;;) {
                doWork();
                Thread.sleep(100);
            }
        }
        catch (InterruptedException e) {
            System.out.println("Worker thread will be interrupted...");
        }
    }

    private void doWork() throws InterruptedException {
        boolean success = queue.offer(PRODUCT, 100, TimeUnit.MILLISECONDS);
        if (success) {
            output.incrementAndGet(); // 可以声明long型的参数获得返回值,作为日志的参数
            // 可以在此处生成记录日志
        }
    }
}
/**
 * 消费者
 */
class Seller implements Runnable {

    private final LinkedBlockingQueue<String> queue;
    private final AtomicLong sales;
    private final CountDownLatch latch;
    private final boolean clear;

    public Seller(LinkedBlockingQueue<String> queue, AtomicLong sales, CountDownLatch latch, boolean clear) {
        this.queue = queue;
        this.sales = sales;
        this.latch = latch;
        this.clear = clear;
    }

    public void run() {
        try {
            latch.await(); // 放闸之前老实的等待着
            for (;;) {
                sale();
                Thread.sleep(500);
            }
        }
        catch (InterruptedException e) {
            if(clear) { // 响应中断请求后,如果有要求则销售完队列的产品后再终止线程
                cleanWarehouse();
            }
            else {
                System.out.println("Seller Thread will be interrupted...");
            }
        }
    }
    private void sale() throws InterruptedException {
        String item = queue.poll(50, TimeUnit.MILLISECONDS);
        if (item != null) {
            sales.incrementAndGet(); // 可以声明long型的参数获得返回值,作为日志的参数
            // 可以在此处生成记录日志
        }
    }

    /**
     * 销售完队列剩余的产品
     */
    private void cleanWarehouse() {
        try {
            while (queue.size() > 0) {
                sale();
            }
        }
        catch (InterruptedException ex) {
            System.out.println("Seller Thread will be interrupted...");
        }
    }
}

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>  
        implements Queue<E>, java.io.Serializable {  
    private static final long serialVersionUID = 196745693267521676L;  
  
    private static class Node<E> {  
        private volatile E item;  
        private volatile Node<E> next;  
  
        private static final  
            AtomicReferenceFieldUpdater<Node, Node>  
            nextUpdater =  
            AtomicReferenceFieldUpdater.newUpdater  
            (Node.class, Node.class, "next");  
        private static final  
            AtomicReferenceFieldUpdater<Node, Object>  
            itemUpdater =  
            AtomicReferenceFieldUpdater.newUpdater  
            (Node.class, Object.class, "item");  
  
        Node(E x) { item = x; }  
  
        Node(E x, Node<E> n) { item = x; next = n; }  
  
        E getItem() {  
            return item;  
        }  
  
        boolean casItem(E cmp, E val) {  
            return itemUpdater.compareAndSet(this, cmp, val);  
        }  
  
        void setItem(E val) {  
            itemUpdater.set(this, val);  
        }  
  
        Node<E> getNext() {  
            return next;  
        }  
  
        boolean casNext(Node<E> cmp, Node<E> val) {  
            return nextUpdater.compareAndSet(this, cmp, val);  
        }  
  
        void setNext(Node<E> val) {  
            nextUpdater.set(this, val);  
        }  
  
    }  
  
    private static final  
        AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>  
        tailUpdater =  
        AtomicReferenceFieldUpdater.newUpdater  
        (ConcurrentLinkedQueue.class, Node.class, "tail");  
    private static final  
        AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>  
        headUpdater =  
        AtomicReferenceFieldUpdater.newUpdater  
        (ConcurrentLinkedQueue.class,  Node.class, "head");  
  
    private boolean casTail(Node<E> cmp, Node<E> val) {  
        return tailUpdater.compareAndSet(this, cmp, val);  
    }  
  
    private boolean casHead(Node<E> cmp, Node<E> val) {  
        return headUpdater.compareAndSet(this, cmp, val);  
    }  
  
    private transient volatile Node<E> head = new Node<E>(null, null);  
  
    private transient volatile Node<E> tail = head;  
    ...  
}

public boolean offer(E e) {  
    if (e == null) throw new NullPointerException();  
    Node<E> n = new Node<E>(e, null);  
    for (;;) {  
        Node<E> t = tail;  
        Node<E> s = t.getNext();  
        if (t == tail) { //------------------------------a  
            if (s == null) { //---------------------------b  
                if (t.casNext(s, n)) { //-------------------c  
                    casTail(t, n); //------------------------d  
                    return true;  
                }  
            } else {  
                casTail(t, s); //----------------------------e  
            }  
        }  
    }  
} 

public final class Counter {  
    private long value = 0;  
    public synchronized long getValue() {  
        return value;  
    }  
    public synchronized long increment() {  
        return ++value;  
    }  
} 

public class NonblockingCounter {  
    private AtomicInteger value;//前面提到过，AtomicInteger类是以原子的方式操作整型变量。  
    public int getValue() {  
        return value.get();  
    }  
    public int increment() {  
        int v;  
        do {  
            v = value.get();  
        while (!value.compareAndSet(v, v + 1));  
        return v + 1;  
    }  
}