Day24 ReentrantLock、并发容器

ReentrantLock

可重入悲观独占锁，与其他两者的比较

特性	`synchronized`	`ReentrantLock`	CAS
实现	JVM 内置（C++ Monitor）	JDK 层（AQS + CAS）	CPU 指令
锁类型	可重入、悲观锁	可重入、悲观锁	乐观锁
阻塞	是（重量级时）	是（可中断）	否（自旋）
功能	自动释放	支持超时、公平锁(默认非公平)	仅原子操作
性能（低竞争）	极优（偏向/轻量级）	略逊	最优
性能（高竞争）	差（重量级）	较好（CLH 队列）	差（自旋）

Lock接口的常用API

加锁后一定要及时释放！！！
- void lock() 获取锁，调用该方法当前线程会获取锁，当锁获得后，该方法返回
- void lockInterruptibly() throws InterruptedException可中断的获取锁，和lock()方法不同之处在于该方法会响应中断，即在锁的获取中可以中断当前线程
- boolean tryLock()尝试非阻塞的获取锁，调用该方法后立即返回。如果能够获取到返回true，否则返回false
- boolean tryLock(long time, TimeUnit unit) throws InterruptedException超时获取锁，当前线程在以下三种情况下会被返回:当前线程在超时时间内获取了锁当前线程在超时时间内被中断超时时间结束，返回false
- void unlock() 释放锁
- Condition newCondition()获取等待通知组件，该组件和当前的锁绑定，当前线程只有获取了锁，才能调用该组件的await()方法，而调用后，当前线程将释放锁

ReentrantLock使用

公平锁与非公平锁

公平锁：线程在获取锁时，按照等待的先后顺序获取锁。
非公平锁：线程在获取锁时，不按照等待的先后顺序获取锁，而是随机获取锁。ReentrantLock默认是非公平锁

▼
java
复制代码
ReentrantLock lock = new ReentrantLock(); //参数默认false，不公平锁  
ReentrantLock lock = new ReentrantLock(true); //公平锁

▼
java
复制代码
class Counter {
    private final ReentrantLock lock = new ReentrantLock(); // 创建 ReentrantLock 对象

    public void recursiveCall(int num) {
        lock.lock(); // 获取锁
        try {
            if (num == 0) {
                return;
            }
            System.out.println("执行递归，num = " + num);
            recursiveCall(num - 1);
        } finally {
            lock.unlock(); // 释放锁
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter(); // 创建计数器对象

        // 测试递归调用
        counter.recursiveCall(10);
    }
}

Condition

Condition提供了线程之间的协调机制的接口，可以将它看作是一个更加灵活、更加强大的wait()和notify()机制，通常与Lock接口（比如ReentrantLock）一起使用。它的核心作用体现在两个方面

等待/通知机制：它允许线程等待某个条件成立，或者通知其他线程某个条件已经满足，这与使用Object的wait()和notify()方法相似，但Condition提供了更高的灵活性和更多的控制。
多条件协调：与每个Object只有一个内置的等待/通知机制不同，一个Lock可以对应多个Condition对象，这意味着可以为不同的等待条件创建不同的Condition，从而实现对多个等待线程集合的独立控制。

核心API

void await() 使当前线程等待，直到被其他线程通过 signal() 或 signalAll() 方法唤醒，或者线程被中断，或者发生了其他不可预知的情况（如假唤醒）。该方法会在等待之前释放当前线程所持有的锁，在被唤醒后会再次尝试获取锁。
boolean await(long time, TimeUnit unit) 使当前线程等待指定的时间，效果同上
void signal()唤醒等待在此 Condition 上的一个线程。如果有多个线程正在等待，则选择其中的一个进行唤醒。被唤醒的线程将从其 await() 调用中返回，并重新尝试获取与此 Condition 关联的锁。
void signalAll()唤醒等待在此 Condition 上的所有线程。

▼
java
复制代码
public class ReentrantLockDemo3 {
    public static void main(String[] args) {
        // 创建队列
        Queue queue = new Queue(5);
        //启动生产者线程
        new Thread(new Producer(queue)).start();
        //启动消费者线程
        new Thread(new Customer(queue)).start();
    }
}

/**
 * 队列封装类
 */
class Queue {
    private Object[] items ;
    int size = 0;
    int takeIndex;
    int putIndex;
    private ReentrantLock lock;
    public Condition notEmpty; //消费者线程阻塞唤醒条件，队列为空阻塞，生产者生产完唤醒
    public Condition notFull; //生产者线程阻塞唤醒条件，队列满了阻塞，消费者消费完唤醒

    public Queue(int capacity){
        this.items = new Object[capacity];
        lock = new ReentrantLock();
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    public void put(Object value) throws Exception {
        //加锁
        lock.lock();
        try {
            while (size == items.length)
                // 队列满了让生产者等待
                notFull.await();

            items[putIndex] = value;
            if (++putIndex == items.length)
                putIndex = 0;
            size++;
            notEmpty.signal(); // 生产完唤醒消费者
        } finally {
            System.out.println("producer生产：" + value);
            //解锁
            lock.unlock();
        }
    }

    public Object take() throws Exception {
        lock.lock();
        try {
            // 队列空了就让消费者等待
            while (size == 0)
                notEmpty.await();

            Object value = items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length)
                takeIndex = 0;
            size--;
            notFull.signal(); //消费完唤醒生产者生产
            return value;
        } finally {
            lock.unlock();
        }
    }
}

/**
 * 生产者
 */
class Producer implements Runnable {
    private Queue queue;
    public Producer(Queue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            // 隔1秒轮询生产一次
            while (true) {
                Thread.sleep(1000);
                queue.put(new Random().nextInt(1000));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

/**
 * 消费者
 */
class Customer implements Runnable {
    private Queue queue;
    public Customer(Queue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            // 隔2秒轮询消费一次
            while (true) {
                Thread.sleep(2000);
                System.out.println("consumer消费：" + queue.take());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

工作原理

基于AQS（AbstractQueuedSynchronizer）+ CAS
- AQS
  - AQS具备特性
    - 阻塞等待队列
    - 共享/独占
    - 公平/非公平
    - 可重入
    - 允许中断
  - 核心结构
    - volatile int state ，值为0时表示无锁，大于0表示已加锁，值表示重入次数。除get/set外，还提供了compareAndSetState()
    - 定义了两种资源访问方式： Exclusive独占和Share共享
  - 主要方法
    - isHeldExclusively()：该线程是否正在独占资源。只有用到condition才需要去实现它。
    - tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。
    - tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。
    - tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
    - tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
  - AQS定义的两种队列
    
    值为0，初始化状态，表示当前节点在sync队列中，等待着获取锁。CANCELLED，值为1，表示当前的线程被取消； SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark； CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中； PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行；
    - 同步等待队列
      
      AQS当中的同步等待队列也称CLH队列，CLH队列是Craig、Landin、Hagersten三人发明的一种基于双向链表数据结构的队列，是FIFO先进先出线程等待队列，Java中的CLH队列是原CLH队列的一个变种,线程由原自旋机制改为阻塞机制。
      
      AQS 依赖CLH同步队列来完成同步状态的管理：
      - 当前线程如果获取同步状态失败时，AQS则会将当前线程已经等待状态等信息构造成一个节点（Node）并将其加入到CLH同步队列，同时会阻塞当前线程
      - 当同步状态释放时，会把首节点唤醒（公平锁），使其再次尝试获取同步状态。
      - 通过signal或signalAll将条件队列中的节点转移到同步队列。（由条件队列转化为同步队列）
    - 条件等待队列
      
      AQS中条件队列是使用单向列表保存的，用nextWaiter来连接:
      - 调用await方法阻塞线程；
      - 当前线程存在于同步队列的头结点，调用await方法进行阻塞（从同步队列转化到条件队列）
  - 基于AQS实现一把独占锁
```
▼
java
复制代码
public class TulingLock extends AbstractQueuedSynchronizer{
    @Override
    protected boolean tryAcquire(int unused) {
        //cas 加锁  state=0
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    @Override
    protected boolean tryRelease(int unused) {
        //释放锁
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock() {
        acquire(1);
    }

    public boolean tryLock() {
        return tryAcquire(1);
    }

    public void unlock() {
        release(1);
    }

    public boolean isLocked() {
        return getState() != 0;
    }
}
```

Semaphore&CountDownLatch&CyclicBarrier

Semaphore 信号量

主要用于在一个时刻允许多个限制数量的线程对共享资源进行并行操作的场景。Semaphore维护了一个计数器，线程可以通过调用acquire()方法来获取Semaphore中的许可证，当计数器为0时，调用acquire()的线程将被阻塞，直到有其他线程释放许可证；线程可以通过调用release()方法来释放Semaphore中的许可证，这会使Semaphore中的计数器增加，从而允许更多的线程访问共享资源。
- 常用API
  
  Semaphore默认是非公平的，允许创建实例时指定是否公平public Semaphore(int permits, boolean fair)
  - acquire() / acquire(int permits) 获取一个/指定数量的许可证，如果获取不到就会一直等待，直到获取到可用许可证或被其他线程中断
  - tryAcquire() / tryAcquire(long timeout, TimeUnit unit) | tryAcquire(int permits) / tryAcquire(int permits, long timeout, TimeUnit unit) 尝试（在指定时间内）获取一个/ 指定数量许可证，获取失败则返回false
  - release() / release(int permits) 释放一个/指定数量许可证，释放后内部空闲许可证计数器会增加
```
▼
java
复制代码
final Semaphore semaphore = new Semaphore(1, true);
// 定义一个线程
new Thread(() -> {
    // 获取许可证
    boolean gotPermit = semaphore.tryAcquire();
    // 如果获取成功就休眠5秒的时间
    if (gotPermit) {
        try {
            System.out.println(Thread.currentThread() + " get one permit.");
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 释放Semaphore的许可证
            // 如果有多个不同线程的许可证，需要慎重考虑锁释放
            // 可能会出现将其他线程许可证被释放的情况
            semaphore.release();
        }
    }
}).start();
// 短暂休眠1秒的时间，确保上面的线程能够启动，并且顺利获取许可证
TimeUnit.SECONDS.sleep(1);
// 主线程在3秒之内肯定是无法获取许可证的，那么主线程将在阻塞3秒之后返回获取许可证失败
if(semaphore.tryAcquire(3, TimeUnit.SECONDS)){
    System.out.println("get the permit");
}else {
    System.out.println("get the permit failure.");
}
```
- 使用场景
  - 接口请求限流
  - 资源池数量限制
CountDownLatch

CountDownLatch（闭锁）是一个同步协助类，可以用于控制一个或多个线程等待多个任务完成后再执行。CountDownLatch 内部维护了一个计数器，该计数器初始值为 N，代表需要等待的线程数目，当一个线程完成了需要等待的任务后，就会调用 countDown() 方法将计数器减 1，当计数器的值为 0 时，等待的线程就会开始执行。
- 常用API
  
  CountDownLatch(int count) cout必须≥0，并且初始化实例时，只能使用一次，用完之后不能再复用。
  - countDown()方法，该方法的主要作用是使得构造CountDownLatch指定的count计数器减一。如果此时CountDownLatch中的计数器已经是0，这种情况下如果再次调用countDown()方法，则会被忽略，也就是说count的值最小只能为0。
  - await() / await（long timeout, TimeUnit unit）方法会使得当前的调用线程进入阻塞状态，直到count为0，其他线程可以将当前线程中断
  - getCount()方法，该方法将返回CountDownLatch当前的计数器数值，该返回值的最小值为0。
```
▼
java
复制代码
public class CountDownLatchDemo2 {
    public static void main(String[] args) throws Exception {

        CountDownLatch countDownLatch = new CountDownLatch(5);
        for (int i = 0; i < 5; i++) {
            final int index = i;
            new Thread(() -> {
                try {
                    Thread.sleep(1000 + ThreadLocalRandom.current().nextInt(2000));
                    System.out.println("任务" + index +"执行完成");
                    countDownLatch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        // 主线程在阻塞，当计数器为0，就唤醒主线程往下执行
        countDownLatch.await();
        System.out.println("主线程:在所有任务运行完成后，进行结果汇总");
    }
}
```
- 使用场景
  - 并行任务汇总同步：协调多个并行任务，完成后再进行下一步操作
  - 资源初始化：等待多个资源初始化完成再使用

CyclicBarrier

CyclicBarrier（循环屏障），是 Java 并发库中的一个同步工具，通过它可以实现让一组线程等待至某个状态（屏障点）之后再全部同时执行。CyclicBarrier可以被重用。CyclicBarrier也适合用于某个串行化任务被分拆成若干个并行执行的子任务，当所有的子任务都执行结束之后再继续接下来的工作。

▼
java
复制代码
 // parties表示屏障拦截的线程数量，每个线程调用 await 方法告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。
 public CyclicBarrier(int parties)
 // 用于在线程到达屏障时，优先执行 barrierAction，方便处理更复杂的业务场景(该线程的执行时机是在到达屏障之后再执行)
 public CyclicBarrier(int parties, Runnable barrierAction)

常用方法

await() / await(long timeout, TimeUnit unit) 调用线程进入阻塞状态，等待指定线程数量全部调用await后再进行下一步操作
reset() 重置循环

▼
java
复制代码
public class CyclicBarrierDemo2 {
    private static int[] getProductsByCategoryId() {
        // 商品列表编号为从1～10的数字
        return IntStream.rangeClosed(1, 10).toArray();
    }
    private static class ProductPrice {
        private final int prodID;
        private double price;

        private ProductPrice(int prodID) {
            this(prodID, -1);
        }
        private ProductPrice(int prodID, double price) {
            this.prodID = prodID;
            this.price = price;
        }
        int getProdID() {
            return prodID;
        }
        void setPrice(double price) {
            this.price = price;
        }
        @Override
        public String toString() {
            return "ProductPrice{" + "prodID=" + prodID + ", price=" + price + '}';
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 根据商品品类获取一组商品ID
        final int[] products = getProductsByCategoryId();
        // 通过转换将商品编号转换为ProductPrice
        List<ProductPrice> list = Arrays.stream(products).mapToObj(ProductPrice::new).collect(toList());
        // 1. 定义CyclicBarrier ，指定parties为子任务数量
        final CyclicBarrier barrier = new CyclicBarrier(list.size());
        // 2.用于存放线程任务的list
        final List<Thread> threadList = new ArrayList<>();
        list.forEach(pp -> {
            Thread thread = new Thread(() -> {
                System.out.println(pp.getProdID() + "开始计算商品价格.");
                try {
                    TimeUnit.SECONDS.sleep(current().nextInt(10));
                    if (pp.prodID % 2 == 0) {
                        pp.setPrice(pp.prodID * 0.9D);
                    } else {
                        pp.setPrice(pp.prodID * 0.71D);
                    }
                    System.out.println(pp.getProdID() + "->价格计算完成.");
                } catch (InterruptedException e) {
                    // ignore exception
                } finally {
                    try {
                        // 3.在此等待其他子线程到达barrier point
                        barrier.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                    }
                }
            });
            threadList.add(thread);
            thread.start();
        });
        // 4. 等待所有子任务线程结束
        threadList.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println("所有价格计算完成.");
        list.forEach(System.out::println);
    }
}

CyclicBarrier 与 CountDownLatch 区别
- CountDownLatch 是一次性的，CyclicBarrier 是可循环利用的
- CoundDownLatch的await方法会等待计数器被count down到0，而执行CyclicBarrier的await方法的线程将会等待其他线程到达barrier point。
- CyclicBarrier内部的计数器count是可被重置的，进而使得CyclicBarrier也可被重复使用，而CoundDownLatch则不能

并发容器

List类

CopyOnWriteArrayList

对应非并发容器ArrayList；用于代替Vector、synchronizedList；利用高并发往往是读多写少的特性，对读操作不加锁，对写操作，先复制一份新的集合，在新的集合上面修改，然后将新集合赋值给旧的引用，并通过volatile 保证其可见性。

应用场景

读多写少的场景
不需要实时更新数据的场景

▼
java
复制代码
public class CopyOnWriteArrayListDemo {

    private static CopyOnWriteArrayList<String> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
    // 模拟初始化的黑名单数据
    static {
        copyOnWriteArrayList.add("ipAddr0");
        copyOnWriteArrayList.add("ipAddr1");
        copyOnWriteArrayList.add("ipAddr2");
    }

    public static void main(String[] args) throws InterruptedException {
        Runnable task = new Runnable() {
            public void run() {
                // 模拟接入用时
                try {
                    Thread.sleep(new Random().nextInt(5000));
                } catch (Exception e) {}

                String currentIP = "ipAddr" + new Random().nextInt(6);
                if (copyOnWriteArrayList.contains(currentIP)) {
                    System.out.println(Thread.currentThread().getName() + " IP " + currentIP + "命中黑名单，拒绝接入处理");
                    return;
                }
                System.out.println(Thread.currentThread().getName() + " IP " + currentIP + "接入处理...");
            }
        };
        new Thread(task, "请求1").start();
        new Thread(task, "请求2").start();
        new Thread(task, "请求3").start();

        new Thread(new Runnable() {
            public void run() {
                // 模拟用时
                try {
                    Thread.sleep(new Random().nextInt(2000));
                } catch (Exception e) {}

                String newBlackIP = "ipAddr3";
                copyOnWriteArrayList.add(newBlackIP);
                System.out.println(Thread.currentThread().getName() + " 添加了新的非法IP " + newBlackIP);
            }
        }, "IP黑名单更新").start();

        Thread.sleep(1000000);
    }
}

优点
- 读操作不加锁，性能高
缺点
- 内存占用问题，每次执行写操作都得拷贝一份，数据量大时对内存压力较大
- 无法保证实时性，写时复制在写操作的时候，内存里会同时有两个对象内存

Map类
- ConcurrentHashMap
  
  对应非并发容器HashMap；用于代替Hashtable、synchronizedMap，支持复合操作作；JDK8之前采用分段锁，JDK8中采用CAS无锁算法+synchronized。
  - JDK1.7 的实现
    
    底层结构用Segments数组+HashEntry数组+链表实现；底层维护一个默认大小为16的Segments数组，每个segment里是一个完整的HashMap，加上一把继承的ReentrantLock，不同线程访问不同的segment时完全不冲突，只有在访问同一segment时才有锁竞争，并发度最高为segment数组的大小。调用size方法时，需要把所有的Segment锁起来计算累加值，慢。
  - JDK1.8实现
    
    移除了Segment，锁力度细化到每个槽位，数据结构与HashMap一致，变成数组+链表+红黑树。插入时先用CAS无锁尝试插入到数组位置，冲突了才用synchronized，而且只锁链表或树的头节点，其他线程照样可操作别的bucket。当链表节点数大于8且数组长度大于等于64时转换为红黑树；
    
    当树中节点数小于6时退化成链表，中间留个缓存，避免频繁转换
- ConcurrentSkipListMap
  
  对应非并发容器TreeMap；用于代替synchronizedSortedMap；用跳表替代平衡树，默认按照key升序。适用于需要高并发性能、支持有序性和区间查询的场景，能够有效地提高系统的性能和可扩展性。
  
  跳表介绍见 redis → zset底层实现
```
▼
java
复制代码
public class ConcurrentSkipListMapDemo {
    public static void main(String[] args) {
        ConcurrentSkipListMap<Integer, String> map = new ConcurrentSkipListMap<>();
        
        // 添加元素
        map.put(1, "a");
        map.put(3, "c");
        map.put(2, "b");
        map.put(4, "d");
        
        // 获取元素
        String value1 = map.get(2);
        System.out.println(value1); // 输出：b
        
        // 遍历元素
        for (Integer key : map.keySet()) {
            String value = map.get(key);
            System.out.println(key + " : " + value);
        }
        
        // 删除元素
        String value2 = map.remove(3);
        System.out.println(value2); // 输出：c
    }
}
```
Set类
- CopyOnWriteArraySet
  
  对应非并发容器HashSet；用于替代synchronizedSet；基于CopyOnWriteArrayList实现，其唯一的不同是在add时调用的是CopyOnWriteArrayList的addIfAbsent方法，其遍历当前Object数组，如Object数组中已有了当前元素，则直接返回，如果没有则放入Object数组的尾部，并返回。
- ConcurrentSkipListSet 基于ConcurrentSkipListMap实现
Queue类
- BlockingQueue BlockingQueue提供了线程安全的队列访问方式：当阻塞队列插入数据时，如果队列已满，线程将会阻塞等待直到队列非满；从阻塞队列取数据时，如果队列已空，线程将会阻塞等待直到队列非空。
  - 应用场景
    1. 线程池
      
      线程池中的任务队列通常是一个阻塞队列。当任务数超过线程池的容量时，新提交的任务将被放入任务队列中等待执行。线程池中的工作线程从任务队列中取出任务进行处理，如果队列为空，则工作线程会被阻塞，直到队列中有新的任务被提交。
    2. 生产者-消费者模型
      
      在生产者-消费者模型中，生产者向队列中添加元素，消费者从队列中取出元素进行处理。阻塞队列可以很好地解决生产者和消费者之间的并发问题，避免线程间的竞争和冲突。
    3. 消息队列
      
      消息队列使用阻塞队列来存储消息，生产者将消息放入队列中，消费者从队列中取出消息进行处理。消息队列可以实现异步通信，提高系统的吞吐量和响应性能，同时还可以将不同的组件解耦，提高系统的可维护性和可扩展性。
    4. 缓存系统
      
      缓存系统使用阻塞队列来存储缓存数据，当缓存数据被更新时，它会被放入队列中，其他线程可以从队列中取出最新的数据进行使用。使用阻塞队列可以避免并发更新缓存数据时的竞争和冲突。
    5. 并发任务处理
      
      在并发任务处理中，可以将待处理的任务放入阻塞队列中，多个工作线程可以从队列中取出任务进行处理。使用阻塞队列可以避免多个线程同时处理同一个任务的问题，并且可以将任务的提交和执行解耦，提高系统的可维护性和可扩展性。
  - ArrayBlockingQueue
    
    ArrayBlockingQueue是最典型的有界阻塞队列，其内部是用数组存储元素的，初始化时需要指定容量大小，利用 ReentrantLock 实现线程安全。ArrayBlockingQueue可以用于实现数据缓存、限流、生产者-消费者模式等各种应用。在生产者-消费者模型中使用时，如果生产速度和消费速度基本匹配的情况下，使用ArrayBlockingQueue是个不错选择；当如果生产速度远远大于消费速度，则会导致队列填满，大量生产线程被阻塞。
    
    ArrayBlockingQueue使用独占锁ReentrantLock实现线程安全，入队和出队操作使用同一个锁对象，也就是只能有一个线程可以进行入队或者出队操作；这也就意味着生产者和消费者无法并行操作，在高并发场景下会成为性能瓶颈。
  - LinkedBlockingQueue
    
    LinkedBlockingQueue是一个基于链表实现的阻塞队列，默认情况下，该阻塞队列的大小为Integer.MAX_VALUE，由于这个数值特别大，所以 LinkedBlockingQueue 也被称作无界队列，代表它几乎没有界限，队列可以随着元素的添加而动态增长，但是如果没有剩余内存，则队列将抛出OOM错误。所以为了避免队列过大造成机器负载或者内存爆满的情况出现，我们在使用的时候建议手动传一个队列的大小。

fail-fast与fail-safe fail-fast(快速失败) 当使用迭代器对集合进行遍历时，如果有线程对集合中的结构进行修改（比如删除），会立即抛出ConcurrentModificationException，ArrayList、HashMap等采用的是fail-fast。可对操作加锁或用对应的线程安全类替代以解决。 fail-safe(安全失败) 对集合结构的修改都会在一个复制集合上进行，不改变原集合内容，因此不会抛出ConcurrentModificationException；例如 CopyOnWriteArrayList、ConcurrentHashMap等采用的是fail-safe。虽然采用fail-safe的集合类都是线程安全的，但是它们无法保证数据实时性，只能保证数据的最终一致性；另外就是内存占用问题，要通过复制来实现读写分离，因此会占用更多的内存

Semaphore&CountDownLatch&CyclicBarrier

并发容器

目录

目录