public class CopyOnWriteArrayList<E>
    implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
    final transient ReentrantLock lock = new ReentrantLock();
    private transient volatile Object[] array;

    // 添加元素，有鎖
    public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock(); // 修改時加鎖，保證并發(fā)安全
        try {
            Object[] elements = getArray(); // 當(dāng)前數(shù)組
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1); // 創(chuàng)建一個新數(shù)組，比老的大一個空間
            newElements[len] = e; // 要添加的元素放進(jìn)新數(shù)組
            setArray(newElements); // 用新數(shù)組替換原來的數(shù)組
            return true;
        } finally {
            lock.unlock(); // 解鎖
        }
    }

    // 讀元素，不加鎖，因此可能讀取到舊數(shù)據(jù)
    public E get(int index) {
        return get(getArray(), index);
    }
}

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    /**
     * 讀寫共用此鎖，線程間通過下面兩個Condition通信
     * 這兩個Condition和lock有緊密聯(lián)系（就是lock的方法生成的）
     * 類似Object的wait/notify
     */
    final ReentrantLock lock;

    /** 隊列不為空的信號，取數(shù)據(jù)的線程需要關(guān)注 */
    private final Condition notEmpty;

    /** 隊列沒滿的信號，寫數(shù)據(jù)的線程需要關(guān)注 */
    private final Condition notFull;

    // 一直阻塞直到有東西可以拿出來
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    // 在尾部插入一個元素，隊列已滿時等待指定時間，如果還是不能插入則返回
    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 {
            // 循環(huán)等待直到隊列有空閑
            while (count == items.length) {
                if (nanos <= 0)
                    return false;// 等待超時，返回
                // 暫時放出鎖，等待一段時間（可能被提前喚醒并搶到鎖，所以需要循環(huán)判斷條件）
                // 這段時間可能其他線程取走了元素，這樣就有機(jī)會插入了
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(e);//插入一個元素
            return true;
        } finally {
            lock.unlock(); //解鎖
        }
    }

import java.util.concurrent.*;

public class Main {

    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        new Thread(() -> {
            try {
                // 沒有休息，瘋狂寫入
                for (int i = 0; ; i++) {
                    System.out.println('放入: ' + i);
                    queue.put(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }).start();
        new Thread(() -> {
            try {
                // 咸魚模式取數(shù)據(jù)
                while (true) {
                    System.out.println('取出: ' + queue.take());
                    Thread.sleep((long) (Math.random() * 2000));
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }).start();
    }

}

/* 輸出:

放入: 0
取出: 0
放入: 1
取出: 1
放入: 2
取出: 2
放入: 3
取出: 3

*/

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(
        0, // 核心線程為0，沒用的線程都被無情拋棄
        Integer.MAX_VALUE, // 最大線程數(shù)理論上是無限了，還沒到這個值機(jī)器資源就被掏空了
        60L, TimeUnit.SECONDS, // 閑置線程60秒后銷毀
        new SynchronousQueue<Runnable>()); // offer時如果沒有空閑線程取出任務(wù)，則會失敗，線程池就會新建一個線程
}