1234567891011121314151617181920212223242526

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(); // 修改时加锁，保证并发安全        try {            Object[] elements = getArray(); // 当前数组            int len = elements.length;            Object[] newElements = Arrays.copyOf(elements, len + 1); // 创建一个新数组，比老的大一个空间            newElements[len] = e; // 要添加的元素放进新数组            setArray(newElements); // 用新数组替换原来的数组            return true;        } finally {            lock.unlock(); // 解锁        }    }

    // 读元素，不加锁，因此可能读取到旧数据    public E get(int index) {        return get(getArray(), index);    }}

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950

public class ArrayBlockingQueue<E> extends AbstractQueue<E>        implements BlockingQueue<E>, java.io.Serializable {    /**     * 读写共用此锁，线程间通过下面两个Condition通信     * 这两个Condition和lock有紧密联系（就是lock的方法生成的）     * 类似Object的wait/notify     */    final ReentrantLock lock;

    /** 队列不为空的信号，取数据的线程需要关注 */    private final Condition notEmpty;

    /** 队列没满的信号，写数据的线程需要关注 */    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 {            // 循环等待直到队列有空闲            while (count == items.length) {                if (nanos <= 0)                    return false;// 等待超时，返回                // 暂时放出锁，等待一段时间（可能被提前唤醒并抢到锁，所以需要循环判断条件）                // 这段时间可能其他线程取走了元素，这样就有机会插入了                nanos = notFull.awaitNanos(nanos);            }            enqueue(e);//插入一个元素            return true;        } finally {            lock.unlock(); //解锁        }    }

12345678910111213141516171819202122232425262728293031323334353637383940414243444546

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 {                // 咸鱼模式取数据                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

*/

1234567

public static ExecutorService newCachedThreadPool() {    return new ThreadPoolExecutor(        0, // 核心线程为0，没用的线程都被无情抛弃        Integer.MAX_VALUE, // 最大线程数理论上是无限了，还没到这个值机器资源就被掏空了        60L, TimeUnit.SECONDS, // 闲置线程60秒后销毁        new SynchronousQueue<Runnable>()); // offer时如果没有空闲线程取出任务，则会失败，线程池就会新建一个线程}