14 параллельных контейнеров Java, сколько вы использовали?

предисловие

Введение в параллельные контейнеры

1. ConcurrentHashMap: Параллельная HashMap
2. CopyOnWriteArrayList: Параллельная версия ArrayList
3. CopyOnWriteArraySet: параллельный набор
4. ConcurrentLinkedQueue: параллельная очередь (на основе связанного списка)
5. ConcurrentLinkedDeque: параллельная очередь (на основе двусвязного списка)
6. ConcurrentSkipListMap: параллельная карта на основе списка пропуска
7. ConcurrentSkipListSet: параллельный набор на основе списка пропуска
8. ArrayBlockingQueue: очередь блокировки (на основе массива)
9. LinkedBlockingQueue: очередь блокировки (на основе связанного списка)
10. LinkedBlockingDeque: очередь блокировки (на основе двусвязного списка)
11. PriorityBlockingQueue: потокобезопасная приоритетная очередь
12. SynchronousQueue: чтение и запись парных очередей
13. LinkedTransferQueue: очередь обмена данными на основе связанного списка
14. DelayQueue: очередь задержки

2.CopyOnWriteArrayList параллельной версии ArrayList

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

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

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

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

Суммировать

Наконец

Прошу всех обратить внимание на мой официальный аккаунт [Программист в погоне за ветром], в нем будут обновляться статьи, а также размещаться отсортированная информация.