AI 摘要 文章对比分析了 Java 7 与 Java 8 的 ConcurrentHashMap 源码：Java 7 采用固定 16 段 Segment 加分段锁，扩容两倍，链表头插；Java 8 改用 Node 数组+链表/红黑树，以 CAS+synchronized 保证并发，扩容时容量翻倍并可能树化，get 均通过 hash 定位再遍历或树查找。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: ConcurrentHashMap 源码分析category: Javatag:Java集合ConcurrentHashMap 作为线程安全的 HashMap ，它的使用频率也是很高。那么它的存储结构和实现原理是怎么样的呢？1. ConcurrentHashMap 1.71. 存储结构Java 7 中 ConcurrentHashMap 的存储结构如上图，ConcurrnetHashMap 由很多个 Segment 组合，而每一个 Segment 是一个类似于 HashMap 的结构，所以每一个 HashMap 的内部可以进行扩容。但是 Segment 的个数一旦初始化就不能改变，默认 Segment 的个数是 16 个，你也可以认为 ConcurrentHashMap 默认支持最多 16 个线程并发。2. 初始化通过 ConcurrentHashMap 的无参构造探寻 ConcurrentHashMap 的初始化流程。 /** * Creates a new, empty map with a default initial capacity (16), * load factor (0.75) and concurrencyLevel (16). */ public ConcurrentHashMap() { this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); }无参构造中调用了有参构造，传入了三个参数的默认值，他们的值是。 /** * 默认初始化容量 */ static final int DEFAULT_INITIAL_CAPACITY = 16; /** * 默认负载因子 */ static final float DEFAULT_LOAD_FACTOR = 0.75f; /** * 默认并发级别 */ static final int DEFAULT_CONCURRENCY_LEVEL = 16;接着看下这个有参构造函数的内部实现逻辑。@SuppressWarnings("unchecked") public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) { // 参数校验 if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); // 校验并发级别大小，大于 1<<16，重置为 65536 if (concurrencyLevel > MAX_SEGMENTS) concurrencyLevel = MAX_SEGMENTS; // Find power-of-two sizes best matching arguments // 2的多少次方 int sshift = 0; int ssize = 1; // 这个循环可以找到 concurrencyLevel 之上最近的 2的次方值 while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } // 记录段偏移量 this.segmentShift = 32 - sshift; // 记录段掩码 this.segmentMask = ssize - 1; // 设置容量 if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; // c = 容量 / ssize ，默认 16 / 16 = 1，这里是计算每个 Segment 中的类似于 HashMap 的容量 int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; int cap = MIN_SEGMENT_TABLE_CAPACITY; //Segment 中的类似于 HashMap 的容量至少是2或者2的倍数 while (cap < c) cap <<= 1; // create segments and segments[0] // 创建 Segment 数组，设置 segments[0] Segment<K,V> s0 = new Segment<K,V>(loadFactor, (int)(cap * loadFactor), (HashEntry<K,V>[])new HashEntry[cap]); Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize]; UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] this.segments = ss; }总结一下在 Java 7 中 ConcurrentHashMap 的初始化逻辑。必要参数校验。校验并发级别 concurrencyLevel 大小，如果大于最大值，重置为最大值。无参构造默认值是 16.寻找并发级别 concurrencyLevel 之上最近的 2 的幂次方值，作为初始化容量大小，默认是 16。记录 segmentShift 偏移量，这个值为【容量 = 2 的 N 次方】中的 N，在后面 Put 时计算位置时会用到。默认是 32 - sshift = 28.记录 segmentMask，默认是 ssize - 1 = 16 -1 = 15.初始化 segments[0]，默认大小为 2，负载因子 0.75，扩容阀值是 2*0.75=1.5，插入第二个值时才会进行扩容。3. put接着上面的初始化参数继续查看 put 方法源码。/** * Maps the specified key to the specified value in this table. * Neither the key nor the value can be null. * * The value can be retrieved by calling the get method * with a key that is equal to the original key. * * @param key key with which the specified value is to be associated * @param value value to be associated with the specified key * @return the previous value associated with key , or * null if there was no mapping for key * @throws NullPointerException if the specified key or value is null */ public V put(K key, V value) { Segment<K,V> s; if (value == null) throw new NullPointerException(); int hash = hash(key); // hash 值无符号右移 28位（初始化时获得），然后与 segmentMask=15 做与运算 // 其实也就是把高4位与segmentMask（1111）做与运算 int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment // 如果查找到的 Segment 为空，初始化 s = ensureSegment(j); return s.put(key, hash, value, false); } /** * Returns the segment for the given index, creating it and * recording in segment table (via CAS) if not already present. * * @param k the index * @return the segment */ @SuppressWarnings("unchecked") private Segment<K,V> ensureSegment(int k) { final Segment<K,V>[] ss = this.segments; long u = (k << SSHIFT) + SBASE; // raw offset Segment<K,V> seg; // 判断 u 位置的 Segment 是否为null if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { Segment<K,V> proto = ss[0]; // use segment 0 as prototype // 获取0号 segment 里的 HashEntry<K,V> 初始化长度 int cap = proto.table.length; // 获取0号 segment 里的 hash 表里的扩容负载因子，所有的 segment 的 loadFactor 是相同的 float lf = proto.loadFactor; // 计算扩容阀值 int threshold = (int)(cap * lf); // 创建一个 cap 容量的 HashEntry 数组 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck // 再次检查 u 位置的 Segment 是否为null，因为这时可能有其他线程进行了操作 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); // 自旋检查 u 位置的 Segment 是否为null while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 使用CAS 赋值，只会成功一次 if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }上面的源码分析了 ConcurrentHashMap 在 put 一个数据时的处理流程，下面梳理下具体流程。计算要 put 的 key 的位置，获取指定位置的 Segment。如果指定位置的 Segment 为空，则初始化这个 Segment.初始化 Segment 流程：检查计算得到的位置的 Segment 是否为 null.为 null 继续初始化，使用 Segment[0] 的容量和负载因子创建一个 HashEntry 数组。再次检查计算得到的指定位置的 Segment 是否为 null.使用创建的 HashEntry 数组初始化这个 Segment.自旋判断计算得到的指定位置的 Segment 是否为 null，使用 CAS 在这个位置赋值为 Segment.Segment.put 插入 key,value 值。上面探究了获取 Segment 段和初始化 Segment 段的操作。最后一行的 Segment 的 put 方法还没有查看，继续分析。final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 获取 ReentrantLock 独占锁，获取不到，scanAndLockForPut 获取。 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry<K,V>[] tab = table; // 计算要put的数据位置 int index = (tab.length - 1) & hash; // CAS 获取 index 坐标的值 HashEntry<K,V> first = entryAt(tab, index); for (HashEntry<K,V> e = first;;) { if (e != null) { // 检查是否 key 已经存在，如果存在，则遍历链表寻找位置，找到后替换 value K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { e.value = value; ++modCount; } break; } e = e.next; } else { // first 有值没说明 index 位置已经有值了，有冲突，链表头插法。 if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; // 容量大于扩容阀值，小于最大容量，进行扩容 if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else // index 位置赋值 node，node 可能是一个元素，也可能是一个链表的表头 setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { unlock(); } return oldValue; }由于 Segment 继承了 ReentrantLock，所以 Segment 内部可以很方便的获取锁，put 流程就用到了这个功能。tryLock() 获取锁，获取不到使用 scanAndLockForPut 方法继续获取。计算 put 的数据要放入的 index 位置，然后获取这个位置上的 HashEntry 。遍历 put 新元素，为什么要遍历？因为这里获取的 HashEntry 可能是一个空元素，也可能是链表已存在，所以要区别对待。如果这个位置上的 HashEntry 不存在：如果当前容量大于扩容阀值，小于最大容量，进行扩容。直接头插法插入。如果这个位置上的 HashEntry 存在：判断链表当前元素 key 和 hash 值是否和要 put 的 key 和 hash 值一致。一致则替换值不一致，获取链表下一个节点，直到发现相同进行值替换，或者链表表里完毕没有相同的。如果当前容量大于扩容阀值，小于最大容量，进行扩容。直接链表头插法插入。如果要插入的位置之前已经存在，替换后返回旧值，否则返回 null.这里面的第一步中的 scanAndLockForPut 操作这里没有介绍，这个方法做的操作就是不断的自旋 tryLock() 获取锁。当自旋次数大于指定次数时，使用 lock() 阻塞获取锁。在自旋时顺表获取下 hash 位置的 HashEntry。private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; // negative while locating node // 自旋获取锁 while (!tryLock()) { HashEntry<K,V> f; // to recheck first below if (retries < 0) { if (e == null) { if (node == null) // speculatively create node node = new HashEntry<K,V>(hash, key, value, null); retries = 0; } else if (key.equals(e.key)) retries = 0; else e = e.next; } else if (++retries > MAX_SCAN_RETRIES) { // 自旋达到指定次数后，阻塞等到只到获取到锁 lock(); break; } else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) { e = first = f; // re-traverse if entry changed retries = -1; } } return node; } 4. 扩容 rehashConcurrentHashMap 的扩容只会扩容到原来的两倍。老数组里的数据移动到新的数组时，位置要么不变，要么变为 index+ oldSize，参数里的 node 会在扩容之后使用链表头插法插入到指定位置。private void rehash(HashEntry<K,V> node) { HashEntry<K,V>[] oldTable = table; // 老容量 int oldCapacity = oldTable.length; // 新容量，扩大两倍 int newCapacity = oldCapacity << 1; // 新的扩容阀值 threshold = (int)(newCapacity * loadFactor); // 创建新的数组 HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 新的掩码，默认2扩容后是4，-1是3，二进制就是11。 int sizeMask = newCapacity - 1; for (int i = 0; i < oldCapacity ; i++) { // 遍历老数组 HashEntry<K,V> e = oldTable[i]; if (e != null) { HashEntry<K,V> next = e.next; // 计算新的位置，新的位置只可能是不便或者是老的位置+老的容量。 int idx = e.hash & sizeMask; if (next == null) // Single node on list // 如果当前位置还不是链表，只是一个元素，直接赋值 newTable[idx] = e; else { // Reuse consecutive sequence at same slot // 如果是链表了 HashEntry<K,V> lastRun = e; int lastIdx = idx; // 新的位置只可能是不便或者是老的位置+老的容量。 // 遍历结束后，lastRun 后面的元素位置都是相同的 for (HashEntry<K,V> last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } // ，lastRun 后面的元素位置都是相同的，直接作为链表赋值到新位置。 newTable[lastIdx] = lastRun; // Clone remaining nodes for (HashEntry<K,V> p = e; p != lastRun; p = p.next) { // 遍历剩余元素，头插法到指定 k 位置。 V v = p.value; int h = p.hash; int k = h & sizeMask; HashEntry<K,V> n = newTable[k]; newTable[k] = new HashEntry<K,V>(h, p.key, v, n); } } } } // 头插法插入新的节点 int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; }有些同学可能会对最后的两个 for 循环有疑惑，这里第一个 for 是为了寻找这样一个节点，这个节点后面的所有 next 节点的新位置都是相同的。然后把这个作为一个链表赋值到新位置。第二个 for 循环是为了把剩余的元素通过头插法插入到指定位置链表。这样实现的原因可能是基于概率统计，有深入研究的同学可以发表下意见。5. get到这里就很简单了，get 方法只需要两步即可。计算得到 key 的存放位置。遍历指定位置查找相同 key 的 value 值。public V get(Object key) { Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 计算得到 key 的存放位置 if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); e != null; e = e.next) { // 如果是链表，遍历查找到相同 key 的 value。 K k; if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; }2. ConcurrentHashMap 1.81. 存储结构可以发现 Java8 的 ConcurrentHashMap 相对于 Java7 来说变化比较大，不再是之前的 Segment 数组 + HashEntry 数组 + 链表，而是 Node 数组 + 链表 / 红黑树。当冲突链表达到一定长度时，链表会转换成红黑树。2. 初始化 initTable/** * Initializes table, using the size recorded in sizeCtl. */ private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { // 如果 sizeCtl < 0 ,说明另外的线程执行CAS 成功，正在进行初始化。 if ((sc = sizeCtl) < 0) // 让出 CPU 使用权 Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = tab = nt; sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; }从源码中可以发现 ConcurrentHashMap 的初始化是通过自旋和 CAS 操作完成的。里面需要注意的是变量 sizeCtl （sizeControl 的缩写），它的值决定着当前的初始化状态。-1 说明正在初始化，其他线程需要自旋等待-N 说明 table 正在进行扩容，高 16 位表示扩容的标识戳，低 16 位减 1 为正在进行扩容的线程数0 表示 table 初始化大小，如果 table 没有初始化\>0 表示 table 扩容的阈值，如果 table 已经初始化。3. put直接过一遍 put 源码。public V put(K key, V value) { return putVal(key, value, false); } /** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { // key 和 value 不能为空 if (key == null || value == null) throw new NullPointerException(); int hash = spread(key.hashCode()); int binCount = 0; for (Node<K,V>[] tab = table;;) { // f = 目标位置元素 Node<K,V> f; int n, i, fh;// fh 后面存放目标位置的元素 hash 值 if (tab == null || (n = tab.length) == 0) // 数组桶为空，初始化数组桶（自旋+CAS) tab = initTable(); else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 桶内为空，CAS 放入，不加锁，成功了就直接 break 跳出 if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; // 使用 synchronized 加锁加入节点 synchronized (f) { if (tabAt(tab, i) == f) { // 说明是链表 if (fh >= 0) { binCount = 1; // 循环加入新的或者覆盖节点 for (Node<K,V> e = f;; ++binCount) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node<K,V> pred = e; if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } else if (f instanceof TreeBin) { // 红黑树 Node<K,V> p; binCount = 2; if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null; }根据 key 计算出 hashcode 。判断是否需要进行初始化。即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。如果都不满足，则利用 synchronized 锁写入数据。如果数量大于 TREEIFY_THRESHOLD 则要执行树化方法，在 treeifyBin 中会首先判断当前数组长度 ≥64 时才会将链表转换为红黑树。4. getget 流程比较简单，直接过一遍源码。public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // key 所在的 hash 位置 int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { // 如果指定位置元素存在，头结点hash值相同 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) // key hash 值相等，key值相同，直接返回元素 value return e.val; } else if (eh < 0) // 头结点hash值小于0，说明正在扩容或者是红黑树，find查找 return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) { // 是链表，遍历查找 if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }总结一下 get 过程：根据 hash 值计算位置。查找到指定位置，如果头节点就是要找的，直接返回它的 value.如果头节点 hash 值小于 0 ，说明正在扩容或者是红黑树，查找之。如果是链表，遍历查找之。总结：总的来说 ConcurrentHashMap 在 Java8 中相对于 Java7 来说变化还是挺大的，3. 总结Java7 中 ConcurrentHashMap 使用的分段锁，也就是每一个 Segment 上同时只有一个线程可以操作，每一个 Segment 都是一个类似 HashMap 数组的结构，它可以扩容，它的冲突会转化为链表。但是 Segment 的个数一但初始化就不能改变。Java8 中的 ConcurrentHashMap 使用的 Synchronized 锁加 CAS 的机制。结构也由 Java7 中的 Segment 数组 + HashEntry 数组 + 链表 进化成了 Node 数组 + 链表 / 红黑树，Node 是类似于一个 HashEntry 的结构。它的冲突再达到一定大小时会转化成红黑树，在冲突小于一定数量时又退回链表。有些同学可能对 Synchronized 的性能存在疑问，其实 Synchronized 锁自从引入锁升级策略后，性能不再是问题，有兴趣的同学可以自己了解下 Synchronized 的锁升级。
AI 摘要 文章梳理了Java并发包中的核心容器：ConcurrentHashMap、CopyOnWriteArrayList、ConcurrentLinkedQueue、BlockingQueue及ConcurrentSkipListMap，分别阐述其线程安全机制、适用场景与性能特点，并对比JDK7/8实现差异，为并发编程选型提供参考。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: Java 常见并发容器总结category: Javatag:Java并发JDK 提供的这些容器大部分在 java.util.concurrent 包中。ConcurrentHashMap : 线程安全的 HashMapCopyOnWriteArrayList : 线程安全的 List，在读多写少的场合性能非常好，远远好于 Vector。ConcurrentLinkedQueue : 高效的并发队列，使用链表实现。可以看做一个线程安全的 LinkedList，这是一个非阻塞队列。BlockingQueue : 这是一个接口，JDK 内部通过链表、数组等方式实现了这个接口。表示阻塞队列，非常适合用于作为数据共享的通道。ConcurrentSkipListMap : 跳表的实现。这是一个 Map，使用跳表的数据结构进行快速查找。ConcurrentHashMap我们知道 HashMap 不是线程安全的，在并发场景下如果要保证一种可行的方式是使用 Collections.synchronizedMap() 方法来包装我们的 HashMap。但这是通过使用一个全局的锁来同步不同线程间的并发访问，因此会带来不可忽视的性能问题。所以就有了 HashMap 的线程安全版本—— ConcurrentHashMap 的诞生。在 JDK1.7 的时候，ConcurrentHashMap 对整个桶数组进行了分割分段(Segment，分段锁)，每一把锁只锁容器其中一部分数据（下面有示意图），多线程访问容器里不同数据段的数据，就不会存在锁竞争，提高并发访问率。到了 JDK1.8 的时候，ConcurrentHashMap 已经摒弃了 Segment 的概念，而是直接用 Node 数组+链表+红黑树的数据结构来实现，并发控制使用 synchronized 和 CAS 来操作。（JDK1.6 以后 synchronized 锁做了很多优化） 整个看起来就像是优化过且线程安全的 HashMap，虽然在 JDK1.8 中还能看到 Segment 的数据结构，但是已经简化了属性，只是为了兼容旧版本。关于 ConcurrentHashMap 的详细介绍，请看我写的这篇文章：ConcurrentHashMap 源码分析。CopyOnWriteArrayList在 JDK1.5 之前，如果想要使用并发安全的 List 只能选择 Vector。而 Vector 是一种老旧的集合，已经被淘汰。Vector 对于增删改查等方法基本都加了 synchronized，这种方式虽然能够保证同步，但这相当于对整个 Vector 加上了一把大锁，使得每个方法执行的时候都要去获得锁，导致性能非常低下。JDK1.5 引入了 Java.util.concurrent（JUC）包，其中提供了很多线程安全且并发性能良好的容器，其中唯一的线程安全 List 实现就是 CopyOnWriteArrayList 。对于大部分业务场景来说，读取操作往往是远大于写入操作的。由于读取操作不会对原有数据进行修改，因此，对于每次读取都进行加锁其实是一种资源浪费。相比之下，我们应该允许多个线程同时访问 List 的内部数据，毕竟对于读取操作来说是安全的。这种思路与 ReentrantReadWriteLock 读写锁的设计思想非常类似，即读读不互斥、读写互斥、写写互斥（只有读读不互斥）。CopyOnWriteArrayList 更进一步地实现了这一思想。为了将读操作性能发挥到极致，CopyOnWriteArrayList 中的读取操作是完全无需加锁的。更加厉害的是，写入操作也不会阻塞读取操作，只有写写才会互斥。这样一来，读操作的性能就可以大幅度提升。CopyOnWriteArrayList 线程安全的核心在于其采用了 写时复制（Copy-On-Write） 的策略，从 CopyOnWriteArrayList 的名字就能看出了。当需要修改（ add，set、remove 等操作） CopyOnWriteArrayList 的内容时，不会直接修改原数组，而是会先创建底层数组的副本，对副本数组进行修改，修改完之后再将修改后的数组赋值回去，这样就可以保证写操作不会影响读操作了。关于 CopyOnWriteArrayList 的详细介绍，请看我写的这篇文章：CopyOnWriteArrayList 源码分析。ConcurrentLinkedQueueJava 提供的线程安全的 Queue 可以分为阻塞队列和非阻塞队列，其中阻塞队列的典型例子是 BlockingQueue，非阻塞队列的典型例子是 ConcurrentLinkedQueue，在实际应用中要根据实际需要选用阻塞队列或者非阻塞队列。 阻塞队列可以通过加锁来实现，非阻塞队列可以通过 CAS 操作实现。从名字可以看出，ConcurrentLinkedQueue这个队列使用链表作为其数据结构．ConcurrentLinkedQueue 应该算是在高并发环境中性能最好的队列了。它之所有能有很好的性能，是因为其内部复杂的实现。ConcurrentLinkedQueue 内部代码我们就不分析了，大家知道 ConcurrentLinkedQueue 主要使用 CAS 非阻塞算法来实现线程安全就好了。ConcurrentLinkedQueue 适合在对性能要求相对较高，同时对队列的读写存在多个线程同时进行的场景，即如果对队列加锁的成本较高则适合使用无锁的 ConcurrentLinkedQueue 来替代。BlockingQueueBlockingQueue 简介上面我们己经提到了 ConcurrentLinkedQueue 作为高性能的非阻塞队列。下面我们要讲到的是阻塞队列——BlockingQueue。阻塞队列（BlockingQueue）被广泛使用在“生产者-消费者”问题中，其原因是 BlockingQueue 提供了可阻塞的插入和移除的方法。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。BlockingQueue 是一个接口，继承自 Queue，所以其实现类也可以作为 Queue 的实现来使用，而 Queue 又继承自 Collection 接口。下面是 BlockingQueue 的相关实现类：下面主要介绍一下 3 个常见的 BlockingQueue 的实现类：ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue 。ArrayBlockingQueueArrayBlockingQueue 是 BlockingQueue 接口的有界队列实现类，底层采用数组来实现。public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, Serializable{}ArrayBlockingQueue 一旦创建，容量不能改变。其并发控制采用可重入锁 ReentrantLock ，不管是插入操作还是读取操作，都需要获取到锁才能进行操作。当队列容量满时，尝试将元素放入队列将导致操作阻塞;尝试从一个空队列中取一个元素也会同样阻塞。ArrayBlockingQueue 默认情况下不能保证线程访问队列的公平性，所谓公平性是指严格按照线程等待的绝对时间顺序，即最先等待的线程能够最先访问到 ArrayBlockingQueue。而非公平性则是指访问 ArrayBlockingQueue 的顺序不是遵守严格的时间顺序，有可能存在，当 ArrayBlockingQueue 可以被访问时，长时间阻塞的线程依然无法访问到 ArrayBlockingQueue。如果保证公平性，通常会降低吞吐量。如果需要获得公平性的 ArrayBlockingQueue，可采用如下代码：private static ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(10,true);LinkedBlockingQueueLinkedBlockingQueue 底层基于单向链表实现的阻塞队列，可以当做无界队列也可以当做有界队列来使用，同样满足 FIFO 的特性，与 ArrayBlockingQueue 相比起来具有更高的吞吐量，为了防止 LinkedBlockingQueue 容量迅速增，损耗大量内存。通常在创建 LinkedBlockingQueue 对象时，会指定其大小，如果未指定，容量等于 Integer.MAX_VALUE 。相关构造方法: /** *某种意义上的无界队列 * Creates a {@code LinkedBlockingQueue} with a capacity of * {@link Integer#MAX_VALUE}. */ public LinkedBlockingQueue() { this(Integer.MAX_VALUE); } /** *有界队列 * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity. * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity} is not greater * than zero */ public LinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; last = head = new Node<E>(null); }PriorityBlockingQueuePriorityBlockingQueue 是一个支持优先级的无界阻塞队列。默认情况下元素采用自然顺序进行排序，也可以通过自定义类实现 compareTo() 方法来指定元素排序规则，或者初始化时通过构造器参数 Comparator 来指定排序规则。PriorityBlockingQueue 并发控制采用的是可重入锁 ReentrantLock，队列为无界队列（ArrayBlockingQueue 是有界队列，LinkedBlockingQueue 也可以通过在构造函数中传入 capacity 指定队列最大的容量，但是 PriorityBlockingQueue 只能指定初始的队列大小，后面插入元素的时候，如果空间不够的话会自动扩容）。简单地说，它就是 PriorityQueue 的线程安全版本。不可以插入 null 值，同时，插入队列的对象必须是可比较大小的（comparable），否则报 ClassCastException 异常。它的插入操作 put 方法不会 block，因为它是无界队列（take 方法在队列为空的时候会阻塞）。推荐文章： 《解读 Java 并发队列 BlockingQueue》ConcurrentSkipListMap下面这部分内容参考了极客时间专栏《数据结构与算法之美》以及《实战 Java 高并发程序设计》。为了引出 ConcurrentSkipListMap，先带着大家简单理解一下跳表。对于一个单链表，即使链表是有序的，如果我们想要在其中查找某个数据，也只能从头到尾遍历链表，这样效率自然就会很低，跳表就不一样了。跳表是一种可以用来快速查找的数据结构，有点类似于平衡树。它们都可以对元素进行快速的查找。但一个重要的区别是：对平衡树的插入和删除往往很可能导致平衡树进行一次全局的调整。而对跳表的插入和删除只需要对整个数据结构的局部进行操作即可。这样带来的好处是：在高并发的情况下，你会需要一个全局锁来保证整个平衡树的线程安全。而对于跳表，你只需要部分锁即可。这样，在高并发环境下，你就可以拥有更好的性能。而就查询的性能而言，跳表的时间复杂度也是 O(logn) 所以在并发数据结构中，JDK 使用跳表来实现一个 Map。跳表的本质是同时维护了多个链表，并且链表是分层的，最低层的链表维护了跳表内所有的元素，每上面一层链表都是下面一层的子集。跳表内的所有链表的元素都是排序的。查找时，可以从顶级链表开始找。一旦发现被查找的元素大于当前链表中的取值，就会转入下一层链表继续找。这也就是说在查找过程中，搜索是跳跃式的。如上图所示，在跳表中查找元素 18。查找 18 的时候原来需要遍历 18 次，现在只需要 7 次即可。针对链表长度比较大的时候，构建索引查找效率的提升就会非常明显。从上面很容易看出，跳表是一种利用空间换时间的算法。使用跳表实现 Map 和使用哈希算法实现 Map 的另外一个不同之处是：哈希并不会保存元素的顺序，而跳表内所有的元素都是排序的。因此在对跳表进行遍历时，你会得到一个有序的结果。所以，如果你的应用需要有序性，那么跳表就是你不二的选择。JDK 中实现这一数据结构的类是 ConcurrentSkipListMap。参考《实战 Java 高并发程序设计》https://javadoop.com/post/java-concurrent-queuehttps://juejin.im/post/5aeebd02518825672f19c546
AI 摘要 AQS是Java并发包核心抽象队列同步器，用volatile int state与CLH队列管理锁状态，支持独占/共享两种模式。ReentrantLock、Semaphore、CountDownLatch等工具均基于其模板方法模式实现。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: AQS 详解category: Javatag:Java并发AQS 介绍AQS 的全称为 AbstractQueuedSynchronizer ，翻译过来的意思就是抽象队列同步器。这个类在 java.util.concurrent.locks 包下面。AQS 就是一个抽象类，主要用来构建锁和同步器。public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { }AQS 为构建锁和同步器提供了一些通用功能的实现，因此，使用 AQS 能简单且高效地构造出应用广泛的大量的同步器，比如我们提到的 ReentrantLock，Semaphore，其他的诸如 ReentrantReadWriteLock，SynchronousQueue等等皆是基于 AQS 的。AQS 原理在面试中被问到并发知识的时候，大多都会被问到“请你说一下自己对于 AQS 原理的理解”。下面给大家一个示例供大家参考，面试不是背题，大家一定要加入自己的思想，即使加入不了自己的思想也要保证自己能够通俗的讲出来而不是背出来。AQS 核心思想AQS 核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态。如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制 AQS 是基于 CLH 锁 （Craig, Landin, and Hagersten locks） 实现的。CLH 锁是对自旋锁的一种改进，是一个虚拟的双向队列（虚拟的双向队列即不存在队列实例，仅存在结点之间的关联关系），暂时获取不到锁的线程将被加入到该队列中。AQS 将每条请求共享资源的线程封装成一个 CLH 队列锁的一个结点（Node）来实现锁的分配。在 CLH 队列锁中，一个节点表示一个线程，它保存着线程的引用（thread）、 当前节点在队列中的状态（waitStatus）、前驱节点（prev）、后继节点（next）。CLH 队列结构如下图所示：关于 AQS 核心数据结构-CLH 锁的详细解读，强烈推荐阅读 Java AQS 核心数据结构-CLH 锁 - Qunar 技术沙龙 这篇文章。AQS(AbstractQueuedSynchronizer)的核心原理图：AQS 使用 int 成员变量 state 表示同步状态，通过内置的 FIFO 线程等待/等待队列 来完成获取资源线程的排队工作。state 变量由 volatile 修饰，用于展示当前临界资源的获锁情况。// 共享变量，使用volatile修饰保证线程可见性 private volatile int state;另外，状态信息 state 可以通过 protected 类型的getState()、setState()和compareAndSetState() 进行操作。并且，这几个方法都是 final 修饰的，在子类中无法被重写。//返回同步状态的当前值 protected final int getState() { return state; } // 设置同步状态的值 protected final void setState(int newState) { state = newState; } //原子地（CAS操作）将同步状态值设置为给定值update如果当前同步状态的值等于expect（期望值） protected final boolean compareAndSetState(int expect, int update) { return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }以可重入的互斥锁 ReentrantLock 为例，它的内部维护了一个 state 变量，用来表示锁的占用状态。state 的初始值为 0，表示锁处于未锁定状态。当线程 A 调用 lock() 方法时，会尝试通过 tryAcquire() 方法独占该锁，并让 state 的值加 1。如果成功了，那么线程 A 就获取到了锁。如果失败了，那么线程 A 就会被加入到一个等待队列（CLH 队列）中，直到其他线程释放该锁。假设线程 A 获取锁成功了，释放锁之前，A 线程自己是可以重复获取此锁的（state 会累加）。这就是可重入性的体现：一个线程可以多次获取同一个锁而不会被阻塞。但是，这也意味着，一个线程必须释放与获取的次数相同的锁，才能让 state 的值回到 0，也就是让锁恢复到未锁定状态。只有这样，其他等待的线程才能有机会获取该锁。线程 A 尝试获取锁的过程如下图所示（图源从 ReentrantLock 的实现看 AQS 的原理及应用 - 美团技术团队）：再以倒计时器 CountDownLatch 以例，任务分为 N 个子线程去执行，state 也初始化为 N（注意 N 要与线程个数一致）。这 N 个子线程开始执行任务，每执行完一个子线程，就调用一次 countDown() 方法。该方法会尝试使用 CAS(Compare and Swap) 操作，让 state 的值减少 1。当所有的子线程都执行完毕后（即 state 的值变为 0），CountDownLatch 会调用 unpark() 方法，唤醒主线程。这时，主线程就可以从 await() 方法（CountDownLatch 中的await() 方法而非 AQS 中的）返回，继续执行后续的操作。AQS 资源共享方式AQS 定义两种资源共享方式：Exclusive（独占，只有一个线程能执行，如ReentrantLock）和Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）。一般来说，自定义同步器的共享方式要么是独占，要么是共享，他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但 AQS 也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。自定义同步器同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用）：使用者继承 AbstractQueuedSynchronizer 并重写指定的方法。将 AQS 组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用。AQS 使用了模板方法模式，自定义同步器时需要重写下面几个 AQS 提供的钩子方法：//独占方式。尝试获取资源，成功则返回true，失败则返回false。 protected boolean tryAcquire(int) //独占方式。尝试释放资源，成功则返回true，失败则返回false。 protected boolean tryRelease(int) //共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。 protected int tryAcquireShared(int) //共享方式。尝试释放资源，成功则返回true，失败则返回false。 protected boolean tryReleaseShared(int) //该线程是否正在独占资源。只有用到condition才需要去实现它。 protected boolean isHeldExclusively()什么是钩子方法呢？ 钩子方法是一种被声明在抽象类中的方法，一般使用 protected 关键字修饰，它可以是空方法（由子类实现），也可以是默认实现的方法。模板设计模式通过钩子方法控制固定步骤的实现。篇幅问题，这里就不详细介绍模板方法模式了，不太了解的小伙伴可以看看这篇文章：用 Java8 改造后的模板方法模式真的是 yyds!。除了上面提到的钩子方法之外，AQS 类中的其他方法都是 final ，所以无法被其他类重写。常见同步工具类下面介绍几个基于 AQS 的常见同步工具类。Semaphore(信号量)介绍synchronized 和 ReentrantLock 都是一次只允许一个线程访问某个资源，而Semaphore(信号量)可以用来控制同时访问特定资源的线程数量。Semaphore 的使用简单，我们这里假设有 N(N>5) 个线程来获取 Semaphore 中的共享资源，下面的代码表示同一时刻 N 个线程中只有 5 个线程能获取到共享资源，其他线程都会阻塞，只有获取到共享资源的线程才能执行。等到有线程释放了共享资源，其他阻塞的线程才能获取到。// 初始共享资源数量 final Semaphore semaphore = new Semaphore(5); // 获取1个许可 semaphore.acquire(); // 释放1个许可 semaphore.release();当初始的资源个数为 1 的时候，Semaphore 退化为排他锁。Semaphore 有两种模式：。公平模式： 调用 acquire() 方法的顺序就是获取许可证的顺序，遵循 FIFO；非公平模式： 抢占式的。Semaphore 对应的两个构造方法如下：public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }这两个构造方法，都必须提供许可的数量，第二个构造方法可以指定是公平模式还是非公平模式，默认非公平模式。Semaphore 通常用于那些资源有明确访问数量限制的场景比如限流（仅限于单机模式，实际项目中推荐使用 Redis +Lua 来做限流）。原理Semaphore 是共享锁的一种实现，它默认构造 AQS 的 state 值为 permits，你可以将 permits 的值理解为许可证的数量，只有拿到许可证的线程才能执行。以无参 acquire 方法为例，调用semaphore.acquire() ，线程尝试获取许可证，如果 state > 0 的话，则表示可以获取成功，如果 state <= 0 的话，则表示许可证数量不足，获取失败。如果可以获取成功的话(state > 0 )，会尝试使用 CAS 操作去修改 state 的值 state=state-1。如果获取失败则会创建一个 Node 节点加入等待队列，挂起当前线程。// 获取1个许可证 public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } // 获取一个或者多个许可证 public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); }acquireSharedInterruptibly方法是 AbstractQueuedSynchronizer 中的默认实现。// 共享模式下获取许可证，获取成功则返回，失败则加入等待队列，挂起线程 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 尝试获取许可证，arg为获取许可证个数，当获取失败时,则创建一个节点加入等待队列，挂起当前线程。 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }这里再以非公平模式（NonfairSync）的为例，看看 tryAcquireShared 方法的实现。// 共享模式下尝试获取资源(在Semaphore中的资源即许可证): protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); } // 非公平的共享模式获取许可证 final int nonfairTryAcquireShared(int acquires) { for (;;) { // 当前可用许可证数量 int available = getState(); /* * 尝试获取许可证，当前可用许可证数量小于等于0时，返回负值，表示获取失败， * 当前可用许可证大于0时才可能获取成功，CAS失败了会循环重新获取最新的值尝试获取 */ int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }以无参 release 方法为例，调用semaphore.release(); ，线程尝试释放许可证，并使用 CAS 操作去修改 state 的值 state=state+1。释放许可证成功之后，同时会唤醒等待队列中的一个线程。被唤醒的线程会重新尝试去修改 state 的值 state=state-1 ，如果 state > 0 则获取令牌成功，否则重新进入等待队列，挂起线程。// 释放一个许可证 public void release() { sync.releaseShared(1); } // 释放一个或者多个许可证 public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); }releaseShared方法是 AbstractQueuedSynchronizer 中的默认实现。// 释放共享锁 // 如果 tryReleaseShared 返回 true，就唤醒等待队列中的一个或多个线程。 public final boolean releaseShared(int arg) { //释放共享锁 if (tryReleaseShared(arg)) { //释放当前节点的后置等待节点 doReleaseShared(); return true; } return false; }tryReleaseShared 方法是Semaphore 的内部类 Sync 重写的一个方法， AbstractQueuedSynchronizer中的默认实现仅仅抛出 UnsupportedOperationException 异常。// 内部类 Sync 中重写的一个方法 // 尝试释放资源 protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); // 可用许可证+1 int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); // CAS修改state的值 if (compareAndSetState(current, next)) return true; } }可以看到，上面提到的几个方法底层基本都是通过同步器 sync 实现的。Sync 是 CountDownLatch 的内部类 , 继承了 AbstractQueuedSynchronizer ，重写了其中的某些方法。并且，Sync 对应的还有两个子类 NonfairSync（对应非公平模式） 和 FairSync（对应公平模式）。private static final class Sync extends AbstractQueuedSynchronizer { // ... } static final class NonfairSync extends Sync { // ... } static final class FairSync extends Sync { // ... }实战public class SemaphoreExample { // 请求的数量 private static final int threadCount = 550; public static void main(String[] args) throws InterruptedException { // 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很慢） ExecutorService threadPool = Executors.newFixedThreadPool(300); // 初始许可证数量 final Semaphore semaphore = new Semaphore(20); for (int i = 0; i < threadCount; i++) { final int threadnum = i; threadPool.execute(() -> {// Lambda 表达式的运用 try { semaphore.acquire();// 获取一个许可，所以可运行线程数量为20/1=20 test(threadnum); semaphore.release();// 释放一个许可 } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); System.out.println("finish"); } public static void test(int threadnum) throws InterruptedException { Thread.sleep(1000);// 模拟请求的耗时操作 System.out.println("threadnum:" + threadnum); Thread.sleep(1000);// 模拟请求的耗时操作 } }执行 acquire() 方法阻塞，直到有一个许可证可以获得然后拿走一个许可证；每个 release 方法增加一个许可证，这可能会释放一个阻塞的 acquire() 方法。然而，其实并没有实际的许可证这个对象，Semaphore 只是维持了一个可获得许可证的数量。 Semaphore 经常用于限制获取某种资源的线程数量。当然一次也可以一次拿取和释放多个许可，不过一般没有必要这样做：semaphore.acquire(5);// 获取5个许可，所以可运行线程数量为20/5=4 test(threadnum); semaphore.release(5);// 释放5个许可除了 acquire() 方法之外，另一个比较常用的与之对应的方法是 tryAcquire() 方法，该方法如果获取不到许可就立即返回 false。issue645 补充内容：Semaphore 与 CountDownLatch 一样，也是共享锁的一种实现。它默认构造 AQS 的 state 为 permits。当执行任务的线程数量超出 permits，那么多余的线程将会被放入等待队列 Park,并自旋判断 state 是否大于 0。只有当 state 大于 0 的时候，阻塞的线程才能继续执行,此时先前执行任务的线程继续执行 release() 方法，release() 方法使得 state 的变量会加 1，那么自旋的线程便会判断成功。如此，每次只有最多不超过 permits 数量的线程能自旋成功，便限制了执行任务线程的数量。CountDownLatch （倒计时器）介绍CountDownLatch 允许 count 个线程阻塞在一个地方，直至所有线程的任务都执行完毕。CountDownLatch 是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当 CountDownLatch 使用完毕后，它不能再次被使用。原理CountDownLatch 是共享锁的一种实现，它默认构造 AQS 的 state 值为 count。这个我们通过 CountDownLatch 的构造方法即可看出。public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } private static final class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); } //... }当线程调用 countDown() 时，其实使用了tryReleaseShared方法以 CAS 的操作来减少 state，直至 state 为 0 。当 state 为 0 时，表示所有的线程都调用了 countDown 方法，那么在 CountDownLatch 上等待的线程就会被唤醒并继续执行。public void countDown() { // Sync 是 CountDownLatch 的内部类 , 继承了 AbstractQueuedSynchronizer sync.releaseShared(1); }releaseShared方法是 AbstractQueuedSynchronizer 中的默认实现。// 释放共享锁 // 如果 tryReleaseShared 返回 true，就唤醒等待队列中的一个或多个线程。 public final boolean releaseShared(int arg) { //释放共享锁 if (tryReleaseShared(arg)) { //释放当前节点的后置等待节点 doReleaseShared(); return true; } return false; }tryReleaseShared 方法是CountDownLatch 的内部类 Sync 重写的一个方法， AbstractQueuedSynchronizer中的默认实现仅仅抛出 UnsupportedOperationException 异常。// 对 state 进行递减，直到 state 变成 0； // 只有 count 递减到 0 时，countDown 才会返回 true protected boolean tryReleaseShared(int releases) { // 自选检查 state 是否为 0 for (;;) { int c = getState(); // 如果 state 已经是 0 了，直接返回 false if (c == 0) return false; // 对 state 进行递减 int nextc = c-1; // CAS 操作更新 state 的值 if (compareAndSetState(c, nextc)) return nextc == 0; } }以无参 await方法为例，当调用 await() 的时候，如果 state 不为 0，那就证明任务还没有执行完毕，await() 就会一直阻塞，也就是说 await() 之后的语句不会被执行（main 线程被加入到等待队列也就是 CLH 队列中了）。然后，CountDownLatch 会自旋 CAS 判断 state == 0，如果 state == 0 的话，就会释放所有等待的线程，await() 方法之后的语句得到执行。// 等待（也可以叫做加锁） public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } // 带有超时时间的等待 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); }acquireSharedInterruptibly方法是 AbstractQueuedSynchronizer 中的默认实现。// 尝试获取锁，获取成功则返回，失败则加入等待队列，挂起线程 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 尝试获得锁，获取成功则返回 if (tryAcquireShared(arg) < 0) // 获取失败加入等待队列，挂起线程 doAcquireSharedInterruptibly(arg); }tryAcquireShared 方法是CountDownLatch 的内部类 Sync 重写的一个方法，其作用就是判断 state 的值是否为 0，是的话就返回 1，否则返回 -1。protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; }实战CountDownLatch 的两种典型用法：某一线程在开始运行前等待 n 个线程执行完毕 : 将 CountDownLatch 的计数器初始化为 n （new CountDownLatch(n)），每当一个任务线程执行完毕，就将计数器减 1 （countdownlatch.countDown()），当计数器的值变为 0 时，在 CountDownLatch 上 await() 的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。实现多个线程开始执行任务的最大并行性：注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的 CountDownLatch 对象，将其计数器初始化为 1 （new CountDownLatch(1)），多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为 0，多个线程同时被唤醒。CountDownLatch 代码示例：public class CountDownLatchExample { // 请求的数量 private static final int THREAD_COUNT = 550; public static void main(String[] args) throws InterruptedException { // 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很慢） // 只是测试使用，实际场景请手动赋值线程池参数 ExecutorService threadPool = Executors.newFixedThreadPool(300); final CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT); for (int i = 0; i < THREAD_COUNT; i++) { final int threadNum = i; threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 表示一个请求已经被完成 countDownLatch.countDown(); } }); } countDownLatch.await(); threadPool.shutdown(); System.out.println("finish"); } public static void test(int threadnum) throws InterruptedException { Thread.sleep(1000); System.out.println("threadNum:" + threadnum); Thread.sleep(1000); } }上面的代码中，我们定义了请求的数量为 550，当这 550 个请求被处理完成之后，才会执行System.out.println("finish");。与 CountDownLatch 的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用 CountDownLatch.await() 方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。其他 N 个线程必须引用闭锁对象，因为他们需要通知 CountDownLatch 对象，他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的；每调用一次这个方法，在构造函数中初始化的 count 值就减 1。所以当 N 个线程都调 用了这个方法，count 的值等于 0，然后主线程就能通过 await()方法，恢复执行自己的任务。再插一嘴：CountDownLatch 的 await() 方法使用不当很容易产生死锁，比如我们上面代码中的 for 循环改为：for (int i = 0; i < threadCount-1; i++) { ....... }这样就导致 count 的值没办法等于 0，然后就会导致一直等待。CyclicBarrier(循环栅栏)介绍CyclicBarrier 和 CountDownLatch 非常类似，它也可以实现线程间的技术等待，但是它的功能比 CountDownLatch 更加复杂和强大。主要应用场景和 CountDownLatch 类似。CountDownLatch 的实现是基于 AQS 的，而 CycliBarrier 是基于 ReentrantLock(ReentrantLock 也属于 AQS 同步器)和 Condition 的。CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是：让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。原理CyclicBarrier 内部通过一个 count 变量作为计数器，count 的初始值为 parties 属性的初始化值，每当一个线程到了栅栏这里了，那么就将计数器减 1。如果 count 值为 0 了，表示这是这一代最后一个线程到达栅栏，就尝试执行我们构造方法中输入的任务。//每次拦截的线程数 private final int parties; //计数器 private int count;下面我们结合源码来简单看看。1、CyclicBarrier 默认的构造方法是 CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用 await() 方法告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。public CyclicBarrier(int parties) { this(parties, null); } public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }其中，parties 就代表了有拦截的线程的数量，当拦截的线程数量达到这个值的时候就打开栅栏，让所有线程通过。2、当调用 CyclicBarrier 对象调用 await() 方法时，实际上调用的是 dowait(false, 0L)方法。 await() 方法就像树立起一个栅栏的行为一样，将线程挡住了，当拦住的线程数量达到 parties 的值时，栅栏才会打开，线程才得以通过执行。public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } }dowait(false, 0L)方法源码分析如下： // 当线程数量或者请求数量达到 count 时 await 之后的方法才会被执行。上面的示例中 count 的值就为 5。 private int count; /** * Main barrier code, covering the various policies. */ private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; // 锁住 lock.lock(); try { final Generation g = generation; if (g.broken) throw new BrokenBarrierException(); // 如果线程中断了，抛出异常 if (Thread.interrupted()) { breakBarrier(); throw new InterruptedException(); } // count 减1 int index = --count; // 当 count 数量减为 0 之后说明最后一个线程已经到达栅栏了，也就是达到了可以执行await 方法之后的条件 if (index == 0) { // tripped boolean ranAction = false; try { final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; // 将 count 重置为 parties 属性的初始化值 // 唤醒之前等待的线程 // 下一波执行开始 nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // loop until tripped, broken, interrupted, or timed out for (;;) { try { if (!timed) trip.await(); else if (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { if (g == generation && ! g.broken) { breakBarrier(); throw ie; } else { // We're about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. Thread.currentThread().interrupt(); } } if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }实战示例 1：public class CyclicBarrierExample1 { // 请求的数量 private static final int threadCount = 550; // 需要同步的线程数量 private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5); public static void main(String[] args) throws InterruptedException { // 创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); for (int i = 0; i < threadCount; i++) { final int threadNum = i; Thread.sleep(1000); threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); } public static void test(int threadnum) throws InterruptedException, BrokenBarrierException { System.out.println("threadnum:" + threadnum + "is ready"); try { /**等待60秒，保证子线程完全执行结束*/ cyclicBarrier.await(60, TimeUnit.SECONDS); } catch (Exception e) { System.out.println("-----CyclicBarrierException------"); } System.out.println("threadnum:" + threadnum + "is finish"); } }运行结果，如下：threadnum:0is ready threadnum:1is ready threadnum:2is ready threadnum:3is ready threadnum:4is ready threadnum:4is finish threadnum:0is finish threadnum:1is finish threadnum:2is finish threadnum:3is finish threadnum:5is ready threadnum:6is ready threadnum:7is ready threadnum:8is ready threadnum:9is ready threadnum:9is finish threadnum:5is finish threadnum:8is finish threadnum:7is finish threadnum:6is finish ......可以看到当线程数量也就是请求数量达到我们定义的 5 个的时候， await() 方法之后的方法才被执行。另外，CyclicBarrier 还提供一个更高级的构造函数 CyclicBarrier(int parties, Runnable barrierAction)，用于在线程到达屏障时，优先执行 barrierAction，方便处理更复杂的业务场景。示例 2：public class CyclicBarrierExample2 { // 请求的数量 private static final int threadCount = 550; // 需要同步的线程数量 private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> { System.out.println("------当线程数达到之后，优先执行------"); }); public static void main(String[] args) throws InterruptedException { // 创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); for (int i = 0; i < threadCount; i++) { final int threadNum = i; Thread.sleep(1000); threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); } public static void test(int threadnum) throws InterruptedException, BrokenBarrierException { System.out.println("threadnum:" + threadnum + "is ready"); cyclicBarrier.await(); System.out.println("threadnum:" + threadnum + "is finish"); } }运行结果，如下：threadnum:0is ready threadnum:1is ready threadnum:2is ready threadnum:3is ready threadnum:4is ready ------当线程数达到之后，优先执行------ threadnum:4is finish threadnum:0is finish threadnum:2is finish threadnum:1is finish threadnum:3is finish threadnum:5is ready threadnum:6is ready threadnum:7is ready threadnum:8is ready threadnum:9is ready ------当线程数达到之后，优先执行------ threadnum:9is finish threadnum:5is finish threadnum:6is finish threadnum:8is finish threadnum:7is finish ......参考Java 并发之 AQS 详解：https://www.cnblogs.com/waterystone/p/4920797.html从 ReentrantLock 的实现看 AQS 的原理及应用：https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html
AI 摘要 CopyOnWriteArrayList 通过“写时复制”实现线程安全，读无锁、写不阻塞读，仅写写互斥，适合读多写少场景；写操作需复制数组，内存与性能开销大，数据弱一致。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: CopyOnWriteArrayList 源码分析category: Javatag:Java集合CopyOnWriteArrayList 简介在 JDK1.5 之前，如果想要使用并发安全的 List 只能选择 Vector。而 Vector 是一种老旧的集合，已经被淘汰。Vector 对于增删改查等方法基本都加了 synchronized，这种方式虽然能够保证同步，但这相当于对整个 Vector 加上了一把大锁，使得每个方法执行的时候都要去获得锁，导致性能非常低下。JDK1.5 引入了 Java.util.concurrent（JUC）包，其中提供了很多线程安全且并发性能良好的容器，其中唯一的线程安全 List 实现就是 CopyOnWriteArrayList 。关于java.util.concurrent 包下常见并发容器的总结，可以看我写的这篇文章：Java 常见并发容器总结 。CopyOnWriteArrayList 到底有什么厉害之处？对于大部分业务场景来说，读取操作往往是远大于写入操作的。由于读取操作不会对原有数据进行修改，因此，对于每次读取都进行加锁其实是一种资源浪费。相比之下，我们应该允许多个线程同时访问 List 的内部数据，毕竟对于读取操作来说是安全的。这种思路与 ReentrantReadWriteLock 读写锁的设计思想非常类似，即读读不互斥、读写互斥、写写互斥（只有读读不互斥）。CopyOnWriteArrayList 更进一步地实现了这一思想。为了将读操作性能发挥到极致，CopyOnWriteArrayList 中的读取操作是完全无需加锁的。更加厉害的是，写入操作也不会阻塞读取操作，只有写写才会互斥。这样一来，读操作的性能就可以大幅度提升。CopyOnWriteArrayList 线程安全的核心在于其采用了 写时复制（Copy-On-Write） 的策略，从 CopyOnWriteArrayList 的名字就能看出了。Copy-On-Write 的思想是什么？CopyOnWriteArrayList名字中的“Copy-On-Write”即写时复制，简称 COW。下面是维基百科对 Copy-On-Write 的介绍，介绍的挺不错：写入时复制（英语：Copy-on-write，简称 COW）是一种计算机程序设计领域的优化策略。其核心思想是，如果有多个调用者（callers）同时请求相同资源（如内存或磁盘上的数据存储），他们会共同获取相同的指针指向相同的资源，直到某个调用者试图修改资源的内容时，系统才会真正复制一份专用副本（private copy）给该调用者，而其他调用者所见到的最初的资源仍然保持不变。这过程对其他的调用者都是透明的。此作法主要的优点是如果调用者没有修改该资源，就不会有副本（private copy）被创建，因此多个调用者只是读取操作时可以共享同一份资源。这里再以 CopyOnWriteArrayList为例介绍：当需要修改（ add，set、remove 等操作） CopyOnWriteArrayList 的内容时，不会直接修改原数组，而是会先创建底层数组的副本，对副本数组进行修改，修改完之后再将修改后的数组赋值回去，这样就可以保证写操作不会影响读操作了。可以看出，写时复制机制非常适合读多写少的并发场景，能够极大地提高系统的并发性能。不过，写时复制机制并不是银弹，其依然存在一些缺点，下面列举几点：内存占用：每次写操作都需要复制一份原始数据，会占用额外的内存空间，在数据量比较大的情况下，可能会导致内存资源不足。写操作开销：每一次写操作都需要复制一份原始数据，然后再进行修改和替换，所以写操作的开销相对较大，在写入比较频繁的场景下，性能可能会受到影响。数据一致性问题：修改操作不会立即反映到最终结果中，还需要等待复制完成，这可能会导致一定的数据一致性问题。……CopyOnWriteArrayList 源码分析这里以 JDK1.8 为例，分析一下 CopyOnWriteArrayList 的底层核心源码。CopyOnWriteArrayList 的类定义如下：public class CopyOnWriteArrayList<E> extends Object implements List<E>, RandomAccess, Cloneable, Serializable { //... }CopyOnWriteArrayList 实现了以下接口：List : 表明它是一个列表，支持添加、删除、查找等操作，并且可以通过下标进行访问。RandomAccess ：这是一个标志接口，表明实现这个接口的 List 集合是支持 快速随机访问 的。Cloneable ：表明它具有拷贝能力，可以进行深拷贝或浅拷贝操作。Serializable : 表明它可以进行序列化操作，也就是可以将对象转换为字节流进行持久化存储或网络传输，非常方便。初始化CopyOnWriteArrayList 中有一个无参构造函数和两个有参构造函数。// 创建一个空的 CopyOnWriteArrayList public CopyOnWriteArrayList() { setArray(new Object[0]); } // 按照集合的迭代器返回的顺序创建一个包含指定集合元素的 CopyOnWriteArrayList public CopyOnWriteArrayList(Collection<? extends E> c) { Object[] elements; if (c.getClass() == CopyOnWriteArrayList.class) elements = ((CopyOnWriteArrayList<?>)c).getArray(); else { elements = c.toArray(); // c.toArray might (incorrectly) not return Object[] (see 6260652) if (elements.getClass() != Object[].class) elements = Arrays.copyOf(elements, elements.length, Object[].class); } setArray(elements); } // 创建一个包含指定数组的副本的列表 public CopyOnWriteArrayList(E[] toCopyIn) { setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object[].class)); }插入元素CopyOnWriteArrayList 的 add()方法有三个版本：add(E e)：在 CopyOnWriteArrayList 的尾部插入元素。add(int index, E element)：在 CopyOnWriteArrayList 的指定位置插入元素。addIfAbsent(E e)：如果指定元素不存在，那么添加该元素。如果成功添加元素则返回 true。这里以add(E e)为例进行介绍：// 插入元素到 CopyOnWriteArrayList 的尾部 public boolean add(E e) { final ReentrantLock lock = this.lock; // 加锁 lock.lock(); try { // 获取原来的数组 Object[] elements = getArray(); // 原来数组的长度 int len = elements.length; // 创建一个长度+1的新数组，并将原来数组的元素复制给新数组 Object[] newElements = Arrays.copyOf(elements, len + 1); // 元素放在新数组末尾 newElements[len] = e; // array指向新数组 setArray(newElements); return true; } finally { // 解锁 lock.unlock(); } }从上面的源码可以看出：add方法内部用到了 ReentrantLock 加锁，保证了同步，避免了多线程写的时候会复制出多个副本出来。锁被修饰保证了锁的内存地址肯定不会被修改，并且，释放锁的逻辑放在 finally 中，可以保证锁能被释放。CopyOnWriteArrayList 通过复制底层数组的方式实现写操作，即先创建一个新的数组来容纳新添加的元素，然后在新数组中进行写操作，最后将新数组赋值给底层数组的引用，替换掉旧的数组。这也就证明了我们前面说的：CopyOnWriteArrayList 线程安全的核心在于其采用了 写时复制（Copy-On-Write） 的策略。每次写操作都需要通过 Arrays.copyOf 复制底层数组，时间复杂度是 O(n) 的，且会占用额外的内存空间。因此，CopyOnWriteArrayList 适用于读多写少的场景，在写操作不频繁且内存资源充足的情况下，可以提升系统的性能表现。CopyOnWriteArrayList 中并没有类似于 ArrayList 的 grow() 方法扩容的操作。Arrays.copyOf 方法的时间复杂度是 O(n)，其中 n 表示需要复制的数组长度。因为这个方法的实现原理是先创建一个新的数组，然后将源数组中的数据复制到新数组中，最后返回新数组。这个方法会复制整个数组，因此其时间复杂度与数组长度成正比，即 O(n)。值得注意的是，由于底层调用了系统级别的拷贝指令，因此在实际应用中这个方法的性能表现比较优秀，但是也需要注意控制复制的数据量，避免出现内存占用过高的情况。读取元素CopyOnWriteArrayList 的读取操作是基于内部数组 array 并没有发生实际的修改，因此在读取操作时不需要进行同步控制和锁操作，可以保证数据的安全性。这种机制下，多个线程可以同时读取列表中的元素。// 底层数组，只能通过getArray和setArray方法访问 private transient volatile Object[] array; public E get(int index) { return get(getArray(), index); } final Object[] getArray() { return array; } private E get(Object[] a, int index) { return (E) a[index]; }不过，get方法是弱一致性的，在某些情况下可能读到旧的元素值。get(int index)方法是分两步进行的：通过getArray()获取当前数组的引用；直接从数组中获取下标为 index 的元素。这个过程并没有加锁，所以在并发环境下可能出现如下情况：线程 1 调用get(int index)方法获取值，内部通过getArray()方法获取到了 array 属性值；线程 2 调用CopyOnWriteArrayList的add、set、remove 等修改方法时，内部通过setArray方法修改了array属性的值；线程 1 还是从旧的 array 数组中取值。获取列表中元素的个数public int size() { return getArray().length; }CopyOnWriteArrayList中的array数组每次复制都刚好能够容纳下所有元素，并不像ArrayList那样会预留一定的空间。因此，CopyOnWriteArrayList中并没有size属性CopyOnWriteArrayList的底层数组的长度就是元素个数，因此size()方法只要返回数组长度就可以了。删除元素CopyOnWriteArrayList删除元素相关的方法一共有 4 个：remove(int index)：移除此列表中指定位置上的元素。将任何后续元素向左移动（从它们的索引中减去 1）。boolean remove(Object o)：删除此列表中首次出现的指定元素，如果不存在该元素则返回 false。boolean removeAll(Collection<?> c)：从此列表中删除指定集合中包含的所有元素。void clear()：移除此列表中的所有元素。这里以remove(int index)为例进行介绍：public E remove(int index) { // 获取可重入锁 final ReentrantLock lock = this.lock; // 加锁 lock.lock(); try { //获取当前array数组 Object[] elements = getArray(); // 获取当前array长度 int len = elements.length; //获取指定索引的元素(旧值) E oldValue = get(elements, index); int numMoved = len - index - 1; // 判断删除的是否是最后一个元素 if (numMoved == 0) // 如果删除的是最后一个元素，直接复制该元素前的所有元素到新的数组 setArray(Arrays.copyOf(elements, len - 1)); else { // 分段复制，将index前的元素和index+1后的元素复制到新数组 // 新数组长度为旧数组长度-1 Object[] newElements = new Object[len - 1]; System.arraycopy(elements, 0, newElements, 0, index); System.arraycopy(elements, index + 1, newElements, index, numMoved); //将新数组赋值给array引用 setArray(newElements); } return oldValue; } finally { // 解锁 lock.unlock(); } }判断元素是否存在CopyOnWriteArrayList提供了两个用于判断指定元素是否在列表中的方法：contains(Object o)：判断是否包含指定元素。containsAll(Collection<?> c)：判断是否保证指定集合的全部元素。// 判断是否包含指定元素 public boolean contains(Object o) { //获取当前array数组 Object[] elements = getArray(); //调用index尝试查找指定元素，如果返回值大于等于0，则返回true，否则返回false return indexOf(o, elements, 0, elements.length) >= 0; } // 判断是否保证指定集合的全部元素 public boolean containsAll(Collection<?> c) { //获取当前array数组 Object[] elements = getArray(); //获取数组长度 int len = elements.length; //遍历指定集合 for (Object e : c) { //循环调用indexOf方法判断，只要有一个没有包含就直接返回false if (indexOf(e, elements, 0, len) < 0) return false; } //最后表示全部包含或者制定集合为空集合，那么返回true return true; }CopyOnWriteArrayList 常用方法测试代码：// 创建一个 CopyOnWriteArrayList 对象 CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>(); // 向列表中添加元素 list.add("Java"); list.add("Python"); list.add("C++"); System.out.println("初始列表：" + list); // 使用 get 方法获取指定位置的元素 System.out.println("列表第二个元素为：" + list.get(1)); // 使用 remove 方法删除指定元素 boolean result = list.remove("C++"); System.out.println("删除结果：" + result); System.out.println("列表删除元素后为：" + list); // 使用 set 方法更新指定位置的元素 list.set(1, "Golang"); System.out.println("列表更新后为：" + list); // 使用 add 方法在指定位置插入元素 list.add(0, "PHP"); System.out.println("列表插入元素后为：" + list); // 使用 size 方法获取列表大小 System.out.println("列表大小为：" + list.size()); // 使用 removeAll 方法删除指定集合中所有出现的元素 result = list.removeAll(List.of("Java", "Golang")); System.out.println("批量删除结果：" + result); System.out.println("列表批量删除元素后为：" + list); // 使用 clear 方法清空列表中所有元素 list.clear(); System.out.println("列表清空后为：" + list);输出：列表更新后为：[Java, Golang] 列表插入元素后为：[PHP, Java, Golang] 列表大小为：3 批量删除结果：true 列表批量删除元素后为：[PHP] 列表清空后为：[]
AI 摘要 文章系统剖析ThreadLocal：从弱引用key、ThreadLocalMap结构、黄金分割哈希、线性探测解决冲突，到set/get、探测式与启发式清理、扩容机制，再延伸到InheritableThreadLocal及分布式链路traceId实战，覆盖原理、源码与项目踩坑。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: ThreadLocal 详解category: Javatag:Java并发前言全文共 10000+字，31 张图，这篇文章同样耗费了不少的时间和精力才创作完成，原创不易，请大家点点关注+在看，感谢。对于ThreadLocal，大家的第一反应可能是很简单呀，线程的变量副本，每个线程隔离。那这里有几个问题大家可以思考一下：ThreadLocal的 key 是弱引用，那么在 ThreadLocal.get()的时候，发生GC之后，key 是否为null？ThreadLocal中ThreadLocalMap的数据结构？ThreadLocalMap的Hash 算法？ThreadLocalMap中Hash 冲突如何解决？ThreadLocalMap的扩容机制？ThreadLocalMap中过期 key 的清理机制？探测式清理和启发式清理流程？ThreadLocalMap.set()方法实现原理？ThreadLocalMap.get()方法实现原理？项目中ThreadLocal使用情况？遇到的坑？……上述的一些问题你是否都已经掌握的很清楚了呢？本文将围绕这些问题使用图文方式来剖析ThreadLocal的点点滴滴。目录注明： 本文源码基于JDK 1.8ThreadLocal代码演示我们先看下ThreadLocal使用示例：public class ThreadLocalTest { private List<String> messages = Lists.newArrayList(); public static final ThreadLocal<ThreadLocalTest> holder = ThreadLocal.withInitial(ThreadLocalTest::new); public static void add(String message) { holder.get().messages.add(message); } public static List<String> clear() { List<String> messages = holder.get().messages; holder.remove(); System.out.println("size: " + holder.get().messages.size()); return messages; } public static void main(String[] args) { ThreadLocalTest.add("一枝花算不算浪漫"); System.out.println(holder.get().messages); ThreadLocalTest.clear(); } }打印结果：[一枝花算不算浪漫] size: 0ThreadLocal对象可以提供线程局部变量，每个线程Thread拥有一份自己的副本变量，多个线程互不干扰。ThreadLocal的数据结构Thread类有一个类型为ThreadLocal.ThreadLocalMap的实例变量threadLocals，也就是说每个线程有一个自己的ThreadLocalMap。ThreadLocalMap有自己的独立实现，可以简单地将它的key视作ThreadLocal，value为代码中放入的值（实际上key并不是ThreadLocal本身，而是它的一个弱引用）。每个线程在往ThreadLocal里放值的时候，都会往自己的ThreadLocalMap里存，读也是以ThreadLocal作为引用，在自己的map里找对应的key，从而实现了线程隔离。ThreadLocalMap有点类似HashMap的结构，只是HashMap是由数组+链表实现的，而ThreadLocalMap中并没有链表结构。我们还要注意Entry， 它的key是ThreadLocal<?> k ，继承自WeakReference， 也就是我们常说的弱引用类型。GC 之后 key 是否为 null？回应开头的那个问题， ThreadLocal 的key是弱引用，那么在ThreadLocal.get()的时候，发生GC之后，key是否是null？为了搞清楚这个问题，我们需要搞清楚Java的四种引用类型：强引用：我们常常 new 出来的对象就是强引用类型，只要强引用存在，垃圾回收器将永远不会回收被引用的对象，哪怕内存不足的时候软引用：使用 SoftReference 修饰的对象被称为软引用，软引用指向的对象在内存要溢出的时候被回收弱引用：使用 WeakReference 修饰的对象被称为弱引用，只要发生垃圾回收，若这个对象只被弱引用指向，那么就会被回收虚引用：虚引用是最弱的引用，在 Java 中使用 PhantomReference 进行定义。虚引用中唯一的作用就是用队列接收对象即将死亡的通知接着再来看下代码，我们使用反射的方式来看看GC后ThreadLocal中的数据情况：(下面代码来源自：https://blog.csdn.net/thewindkee/article/details/103726942 本地运行演示 GC 回收场景)public class ThreadLocalDemo { public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException, InterruptedException { Thread t = new Thread(()->test("abc",false)); t.start(); t.join(); System.out.println("--gc后--"); Thread t2 = new Thread(() -> test("def", true)); t2.start(); t2.join(); } private static void test(String s,boolean isGC) { try { new ThreadLocal<>().set(s); if (isGC) { System.gc(); } Thread t = Thread.currentThread(); Class<? extends Thread> clz = t.getClass(); Field field = clz.getDeclaredField("threadLocals"); field.setAccessible(true); Object ThreadLocalMap = field.get(t); Class<?> tlmClass = ThreadLocalMap.getClass(); Field tableField = tlmClass.getDeclaredField("table"); tableField.setAccessible(true); Object[] arr = (Object[]) tableField.get(ThreadLocalMap); for (Object o : arr) { if (o != null) { Class<?> entryClass = o.getClass(); Field valueField = entryClass.getDeclaredField("value"); Field referenceField = entryClass.getSuperclass().getSuperclass().getDeclaredField("referent"); valueField.setAccessible(true); referenceField.setAccessible(true); System.out.println(String.format("弱引用key:%s,值:%s", referenceField.get(o), valueField.get(o))); } } } catch (Exception e) { e.printStackTrace(); } } }结果如下：弱引用key:java.lang.ThreadLocal@433619b6,值:abc 弱引用key:java.lang.ThreadLocal@418a15e3,值:java.lang.ref.SoftReference@bf97a12 --gc后-- 弱引用key:null,值:def如图所示，因为这里创建的ThreadLocal并没有指向任何值，也就是没有任何引用：new ThreadLocal<>().set(s);所以这里在GC之后，key就会被回收，我们看到上面debug中的referent=null, 如果改动一下代码：这个问题刚开始看，如果没有过多思考，弱引用，还有垃圾回收，那么肯定会觉得是null。其实是不对的，因为题目说的是在做 ThreadLocal.get() 操作，证明其实还是有强引用存在的，所以 key 并不为 null，如下图所示，ThreadLocal的强引用仍然是存在的。如果我们的强引用不存在的话，那么 key 就会被回收，也就是会出现我们 value 没被回收，key 被回收，导致 value 永远存在，出现内存泄漏。ThreadLocal.set()方法源码详解ThreadLocal中的set方法原理如上图所示，很简单，主要是判断ThreadLocalMap是否存在，然后使用ThreadLocal中的set方法进行数据处理。代码如下：public void set(T value) { Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) map.set(this, value); else createMap(t, value); } void createMap(Thread t, T firstValue) { t.threadLocals = new ThreadLocalMap(this, firstValue); }主要的核心逻辑还是在ThreadLocalMap中的，一步步往下看，后面还有更详细的剖析。ThreadLocalMap Hash 算法既然是Map结构，那么ThreadLocalMap当然也要实现自己的hash算法来解决散列表数组冲突问题。int i = key.threadLocalHashCode & (len-1);ThreadLocalMap中hash算法很简单，这里i就是当前 key 在散列表中对应的数组下标位置。这里最关键的就是threadLocalHashCode值的计算，ThreadLocal中有一个属性为HASH_INCREMENT = 0x61c88647public class ThreadLocal<T> { private final int threadLocalHashCode = nextHashCode(); private static AtomicInteger nextHashCode = new AtomicInteger(); private static final int HASH_INCREMENT = 0x61c88647; private static int nextHashCode() { return nextHashCode.getAndAdd(HASH_INCREMENT); } static class ThreadLocalMap { ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) { table = new Entry[INITIAL_CAPACITY]; int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1); table[i] = new Entry(firstKey, firstValue); size = 1; setThreshold(INITIAL_CAPACITY); } } }每当创建一个ThreadLocal对象，这个ThreadLocal.nextHashCode 这个值就会增长 0x61c88647 。这个值很特殊，它是斐波那契数 也叫 黄金分割数。hash增量为 这个数字，带来的好处就是 hash 分布非常均匀。我们自己可以尝试下：可以看到产生的哈希码分布很均匀，这里不去细纠斐波那契具体算法，感兴趣的可以自行查阅相关资料。ThreadLocalMap Hash 冲突注明： 下面所有示例图中，绿色块Entry代表正常数据，灰色块代表Entry的key值为null，已被垃圾回收。白色块表示Entry为null。虽然ThreadLocalMap中使用了黄金分割数来作为hash计算因子，大大减少了Hash冲突的概率，但是仍然会存在冲突。HashMap中解决冲突的方法是在数组上构造一个链表结构，冲突的数据挂载到链表上，如果链表长度超过一定数量则会转化成红黑树。而 ThreadLocalMap 中并没有链表结构，所以这里不能使用 HashMap 解决冲突的方式了。如上图所示，如果我们插入一个value=27的数据，通过 hash 计算后应该落入槽位 4 中，而槽位 4 已经有了 Entry 数据。此时就会线性向后查找，一直找到 Entry 为 null 的槽位才会停止查找，将当前元素放入此槽位中。当然迭代过程中还有其他的情况，比如遇到了 Entry 不为 null 且 key 值相等的情况，还有 Entry 中的 key 值为 null 的情况等等都会有不同的处理，后面会一一详细讲解。这里还画了一个Entry中的key为null的数据（Entry=2 的灰色块数据），因为key值是弱引用类型，所以会有这种数据存在。在set过程中，如果遇到了key过期的Entry数据，实际上是会进行一轮探测式清理操作的，具体操作方式后面会讲到。ThreadLocalMap.set()详解ThreadLocalMap.set()原理图解看完了ThreadLocal hash 算法后，我们再来看set是如何实现的。往ThreadLocalMap中set数据（新增或者更新数据）分为好几种情况，针对不同的情况我们画图来说明。第一种情况： 通过hash计算后的槽位对应的Entry数据为空：这里直接将数据放到该槽位即可。第二种情况： 槽位数据不为空，key值与当前ThreadLocal通过hash计算获取的key值一致：这里直接更新该槽位的数据。第三种情况： 槽位数据不为空，往后遍历过程中，在找到Entry为null的槽位之前，没有遇到key过期的Entry：遍历散列数组，线性往后查找，如果找到Entry为null的槽位，则将数据放入该槽位中，或者往后遍历过程中，遇到了key 值相等的数据，直接更新即可。第四种情况： 槽位数据不为空，往后遍历过程中，在找到Entry为null的槽位之前，遇到key过期的Entry，如下图，往后遍历过程中，遇到了index=7的槽位数据Entry的key=null：散列数组下标为 7 位置对应的Entry数据key为null，表明此数据key值已经被垃圾回收掉了，此时就会执行replaceStaleEntry()方法，该方法含义是替换过期数据的逻辑，以index=7位起点开始遍历，进行探测式数据清理工作。初始化探测式清理过期数据扫描的开始位置：slotToExpunge = staleSlot = 7以当前staleSlot开始 向前迭代查找，找其他过期的数据，然后更新过期数据起始扫描下标slotToExpunge。for循环迭代，直到碰到Entry为null结束。如果找到了过期的数据，继续向前迭代，直到遇到Entry=null的槽位才停止迭代，如下图所示，slotToExpunge 被更新为 0：以当前节点(index=7)向前迭代，检测是否有过期的Entry数据，如果有则更新slotToExpunge值。碰到null则结束探测。以上图为例slotToExpunge被更新为 0。上面向前迭代的操作是为了更新探测清理过期数据的起始下标slotToExpunge的值，这个值在后面会讲解，它是用来判断当前过期槽位staleSlot之前是否还有过期元素。接着开始以staleSlot位置(index=7)向后迭代，如果找到了相同 key 值的 Entry 数据：从当前节点staleSlot向后查找key值相等的Entry元素，找到后更新Entry的值并交换staleSlot元素的位置(staleSlot位置为过期元素)，更新Entry数据，然后开始进行过期Entry的清理工作，如下图所示：向后遍历过程中，如果没有找到相同 key 值的 Entry 数据：从当前节点staleSlot向后查找key值相等的Entry元素，直到Entry为null则停止寻找。通过上图可知，此时table中没有key值相同的Entry。创建新的Entry，替换table[stableSlot]位置：替换完成后也是进行过期元素清理工作，清理工作主要是有两个方法：expungeStaleEntry()和cleanSomeSlots()，具体细节后面会讲到，请继续往后看。ThreadLocalMap.set()源码详解上面已经用图的方式解析了set()实现的原理，其实已经很清晰了，我们接着再看下源码：java.lang.ThreadLocal.ThreadLocalMap.set():private void set(ThreadLocal<?> key, Object value) { Entry[] tab = table; int len = tab.length; int i = key.threadLocalHashCode & (len-1); for (Entry e = tab[i]; e != null; e = tab[i = nextIndex(i, len)]) { ThreadLocal<?> k = e.get(); if (k == key) { e.value = value; return; } if (k == null) { replaceStaleEntry(key, value, i); return; } } tab[i] = new Entry(key, value); int sz = ++size; if (!cleanSomeSlots(i, sz) && sz >= threshold) rehash(); }这里会通过key来计算在散列表中的对应位置，然后以当前key对应的桶的位置向后查找，找到可以使用的桶。Entry[] tab = table; int len = tab.length; int i = key.threadLocalHashCode & (len-1);什么情况下桶才是可以使用的呢？k = key 说明是替换操作，可以使用碰到一个过期的桶，执行替换逻辑，占用过期桶查找过程中，碰到桶中Entry=null的情况，直接使用接着就是执行for循环遍历，向后查找，我们先看下nextIndex()、prevIndex()方法实现：private static int nextIndex(int i, int len) { return ((i + 1 < len) ? i + 1 : 0); } private static int prevIndex(int i, int len) { return ((i - 1 >= 0) ? i - 1 : len - 1); }接着看剩下for循环中的逻辑：遍历当前key值对应的桶中Entry数据为空，这说明散列数组这里没有数据冲突，跳出for循环，直接set数据到对应的桶中如果key值对应的桶中Entry数据不为空 2.1 如果k = key，说明当前set操作是一个替换操作，做替换逻辑，直接返回 2.2 如果key = null，说明当前桶位置的Entry是过期数据，执行replaceStaleEntry()方法(核心方法)，然后返回for循环执行完毕，继续往下执行说明向后迭代的过程中遇到了entry为null的情况 3.1 在Entry为null的桶中创建一个新的Entry对象 3.2 执行++size操作调用cleanSomeSlots()做一次启发式清理工作，清理散列数组中Entry的key过期的数据 4.1 如果清理工作完成后，未清理到任何数据，且size超过了阈值(数组长度的 2/3)，进行rehash()操作 4.2 rehash()中会先进行一轮探测式清理，清理过期key，清理完成后如果size >= threshold - threshold / 4，就会执行真正的扩容逻辑(扩容逻辑往后看)接着重点看下replaceStaleEntry()方法，replaceStaleEntry()方法提供替换过期数据的功能，我们可以对应上面第四种情况的原理图来再回顾下，具体代码如下：java.lang.ThreadLocal.ThreadLocalMap.replaceStaleEntry():private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) { Entry[] tab = table; int len = tab.length; Entry e; int slotToExpunge = staleSlot; for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)) if (e.get() == null) slotToExpunge = i; for (int i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); if (k == key) { e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e; if (slotToExpunge == staleSlot) slotToExpunge = i; cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; } if (k == null && slotToExpunge == staleSlot) slotToExpunge = i; } tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value); if (slotToExpunge != staleSlot) cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); }slotToExpunge表示开始探测式清理过期数据的开始下标，默认从当前的staleSlot开始。以当前的staleSlot开始，向前迭代查找，找到没有过期的数据，for循环一直碰到Entry为null才会结束。如果向前找到了过期数据，更新探测清理过期数据的开始下标为 i，即slotToExpunge=ifor (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)){ if (e.get() == null){ slotToExpunge = i; } }接着开始从staleSlot向后查找，也是碰到Entry为null的桶结束。如果迭代过程中，碰到 k == key，这说明这里是替换逻辑，替换新数据并且交换当前staleSlot位置。如果slotToExpunge == staleSlot，这说明replaceStaleEntry()一开始向前查找过期数据时并未找到过期的Entry数据，接着向后查找过程中也未发现过期数据，修改开始探测式清理过期数据的下标为当前循环的 index，即slotToExpunge = i。最后调用cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);进行启发式过期数据清理。if (k == key) { e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e; if (slotToExpunge == staleSlot) slotToExpunge = i; cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; }cleanSomeSlots()和expungeStaleEntry()方法后面都会细讲，这两个是和清理相关的方法，一个是过期key相关Entry的启发式清理(Heuristically scan)，另一个是过期key相关Entry的探测式清理。如果 k != key则会接着往下走，k == null说明当前遍历的Entry是一个过期数据，slotToExpunge == staleSlot说明，一开始的向前查找数据并未找到过期的Entry。如果条件成立，则更新slotToExpunge 为当前位置，这个前提是前驱节点扫描时未发现过期数据。if (k == null && slotToExpunge == staleSlot) slotToExpunge = i;往后迭代的过程中如果没有找到k == key的数据，且碰到Entry为null的数据，则结束当前的迭代操作。此时说明这里是一个添加的逻辑，将新的数据添加到table[staleSlot] 对应的slot中。tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value);最后判断除了staleSlot以外，还发现了其他过期的slot数据，就要开启清理数据的逻辑：if (slotToExpunge != staleSlot) cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);ThreadLocalMap过期 key 的探测式清理流程上面我们有提及ThreadLocalMap的两种过期key数据清理方式：探测式清理和启发式清理。我们先讲下探测式清理，也就是expungeStaleEntry方法，遍历散列数组，从开始位置向后探测清理过期数据，将过期数据的Entry设置为null，沿途中碰到未过期的数据则将此数据rehash后重新在table数组中定位，如果定位的位置已经有了数据，则会将未过期的数据放到最靠近此位置的Entry=null的桶中，使rehash后的Entry数据距离正确的桶的位置更近一些。操作逻辑如下：如上图，set(27) 经过 hash 计算后应该落到index=4的桶中，由于index=4桶已经有了数据，所以往后迭代最终数据放入到index=7的桶中，放入后一段时间后index=5中的Entry数据key变为了null如果再有其他数据set到map中，就会触发探测式清理操作。如上图，执行探测式清理后，index=5的数据被清理掉，继续往后迭代，到index=7的元素时，经过rehash后发现该元素正确的index=4，而此位置已经有了数据，往后查找离index=4最近的Entry=null的节点(刚被探测式清理掉的数据：index=5)，找到后移动index= 7的数据到index=5中，此时桶的位置离正确的位置index=4更近了。经过一轮探测式清理后，key过期的数据会被清理掉，没过期的数据经过rehash重定位后所处的桶位置理论上更接近i= key.hashCode & (tab.len - 1)的位置。这种优化会提高整个散列表查询性能。接着看下expungeStaleEntry()具体流程，我们还是以先原理图后源码讲解的方式来一步步梳理：我们假设expungeStaleEntry(3) 来调用此方法，如上图所示，我们可以看到ThreadLocalMap中table的数据情况，接着执行清理操作：第一步是清空当前staleSlot位置的数据，index=3位置的Entry变成了null。然后接着往后探测：执行完第二步后，index=4 的元素挪到 index=3 的槽位中。继续往后迭代检查，碰到正常数据，计算该数据位置是否偏移，如果被偏移，则重新计算slot位置，目的是让正常数据尽可能存放在正确位置或离正确位置更近的位置在往后迭代的过程中碰到空的槽位，终止探测，这样一轮探测式清理工作就完成了，接着我们继续看看具体实现源代码：private int expungeStaleEntry(int staleSlot) { Entry[] tab = table; int len = tab.length; tab[staleSlot].value = null; tab[staleSlot] = null; size--; Entry e; int i; for (i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; tab[i] = null; size--; } else { int h = k.threadLocalHashCode & (len - 1); if (h != i) { tab[i] = null; while (tab[h] != null) h = nextIndex(h, len); tab[h] = e; } } } return i; }这里我们还是以staleSlot=3 来做示例说明，首先是将tab[staleSlot]槽位的数据清空，然后设置size--接着以staleSlot位置往后迭代，如果遇到k==null的过期数据，也是清空该槽位数据，然后size--ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; tab[i] = null; size--; }如果key没有过期，重新计算当前key的下标位置是不是当前槽位下标位置，如果不是，那么说明产生了hash冲突，此时以新计算出来正确的槽位位置往后迭代，找到最近一个可以存放entry的位置。int h = k.threadLocalHashCode & (len - 1); if (h != i) { tab[i] = null; while (tab[h] != null) h = nextIndex(h, len); tab[h] = e; }这里是处理正常的产生Hash冲突的数据，经过迭代后，有过Hash冲突数据的Entry位置会更靠近正确位置，这样的话，查询的时候 效率才会更高。ThreadLocalMap扩容机制在ThreadLocalMap.set()方法的最后，如果执行完启发式清理工作后，未清理到任何数据，且当前散列数组中Entry的数量已经达到了列表的扩容阈值(len*2/3)，就开始执行rehash()逻辑：if (!cleanSomeSlots(i, sz) && sz >= threshold) rehash();接着看下rehash()具体实现：private void rehash() { expungeStaleEntries(); if (size >= threshold - threshold / 4) resize(); } private void expungeStaleEntries() { Entry[] tab = table; int len = tab.length; for (int j = 0; j < len; j++) { Entry e = tab[j]; if (e != null && e.get() == null) expungeStaleEntry(j); } }这里首先是会进行探测式清理工作，从table的起始位置往后清理，上面有分析清理的详细流程。清理完成之后，table中可能有一些key为null的Entry数据被清理掉，所以此时通过判断size >= threshold - threshold / 4 也就是size >= threshold * 3/4 来决定是否扩容。我们还记得上面进行rehash()的阈值是size >= threshold，所以当面试官套路我们ThreadLocalMap扩容机制的时候 我们一定要说清楚这两个步骤：接着看看具体的resize()方法，为了方便演示，我们以oldTab.len=8来举例：扩容后的tab的大小为oldLen * 2，然后遍历老的散列表，重新计算hash位置，然后放到新的tab数组中，如果出现hash冲突则往后寻找最近的entry为null的槽位，遍历完成之后，oldTab中所有的entry数据都已经放入到新的tab中了。重新计算tab下次扩容的阈值，具体代码如下：private void resize() { Entry[] oldTab = table; int oldLen = oldTab.length; int newLen = oldLen * 2; Entry[] newTab = new Entry[newLen]; int count = 0; for (int j = 0; j < oldLen; ++j) { Entry e = oldTab[j]; if (e != null) { ThreadLocal<?> k = e.get(); if (k == null) { e.value = null; } else { int h = k.threadLocalHashCode & (newLen - 1); while (newTab[h] != null) h = nextIndex(h, newLen); newTab[h] = e; count++; } } } setThreshold(newLen); size = count; table = newTab; }ThreadLocalMap.get()详解上面已经看完了set()方法的源码，其中包括set数据、清理数据、优化数据桶的位置等操作，接着看看get()操作的原理。ThreadLocalMap.get()图解第一种情况： 通过查找key值计算出散列表中slot位置，然后该slot位置中的Entry.key和查找的key一致，则直接返回：第二种情况： slot位置中的Entry.key和要查找的key不一致：我们以get(ThreadLocal1)为例，通过hash计算后，正确的slot位置应该是 4，而index=4的槽位已经有了数据，且key值不等于ThreadLocal1，所以需要继续往后迭代查找。迭代到index=5的数据时，此时Entry.key=null，触发一次探测式数据回收操作，执行expungeStaleEntry()方法，执行完后，index 5,8的数据都会被回收，而index 6,7的数据都会前移。index 6,7前移之后，继续从 index=5 往后迭代，于是就在 index=6 找到了key值相等的Entry数据，如下图所示：ThreadLocalMap.get()源码详解java.lang.ThreadLocal.ThreadLocalMap.getEntry():private Entry getEntry(ThreadLocal<?> key) { int i = key.threadLocalHashCode & (table.length - 1); Entry e = table[i]; if (e != null && e.get() == key) return e; else return getEntryAfterMiss(key, i, e); } private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) { Entry[] tab = table; int len = tab.length; while (e != null) { ThreadLocal<?> k = e.get(); if (k == key) return e; if (k == null) expungeStaleEntry(i); else i = nextIndex(i, len); e = tab[i]; } return null; }ThreadLocalMap过期 key 的启发式清理流程上面多次提及到ThreadLocalMap过期 key 的两种清理方式：探测式清理(expungeStaleEntry())、启发式清理(cleanSomeSlots())探测式清理是以当前Entry 往后清理，遇到值为null则结束清理，属于线性探测清理。而启发式清理被作者定义为：Heuristically scan some cells looking for stale entries.具体代码如下：private boolean cleanSomeSlots(int i, int n) { boolean removed = false; Entry[] tab = table; int len = tab.length; do { i = nextIndex(i, len); Entry e = tab[i]; if (e != null && e.get() == null) { n = len; removed = true; i = expungeStaleEntry(i); } } while ( (n >>>= 1) != 0); return removed; }InheritableThreadLocal我们使用ThreadLocal的时候，在异步场景下是无法给子线程共享父线程中创建的线程副本数据的。为了解决这个问题，JDK 中还有一个InheritableThreadLocal类，我们来看一个例子：public class InheritableThreadLocalDemo { public static void main(String[] args) { ThreadLocal<String> ThreadLocal = new ThreadLocal<>(); ThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>(); ThreadLocal.set("父类数据:threadLocal"); inheritableThreadLocal.set("父类数据:inheritableThreadLocal"); new Thread(new Runnable() { @Override public void run() { System.out.println("子线程获取父类ThreadLocal数据：" + ThreadLocal.get()); System.out.println("子线程获取父类inheritableThreadLocal数据：" + inheritableThreadLocal.get()); } }).start(); } }打印结果：子线程获取父类ThreadLocal数据：null 子线程获取父类inheritableThreadLocal数据：父类数据:inheritableThreadLocal实现原理是子线程是通过在父线程中通过调用new Thread()方法来创建子线程，Thread#init方法在Thread的构造方法中被调用。在init方法中拷贝父线程数据到子线程中：private void init(ThreadGroup g, Runnable target, String name, long stackSize, AccessControlContext acc, boolean inheritThreadLocals) { if (name == null) { throw new NullPointerException("name cannot be null"); } if (inheritThreadLocals && parent.inheritableThreadLocals != null) this.inheritableThreadLocals = ThreadLocal.createInheritedMap(parent.inheritableThreadLocals); this.stackSize = stackSize; tid = nextThreadID(); }但InheritableThreadLocal仍然有缺陷，一般我们做异步化处理都是使用的线程池，而InheritableThreadLocal是在new Thread中的init()方法给赋值的，而线程池是线程复用的逻辑，所以这里会存在问题。当然，有问题出现就会有解决问题的方案，阿里巴巴开源了一个TransmittableThreadLocal组件就可以解决这个问题，这里就不再延伸，感兴趣的可自行查阅资料。ThreadLocal项目中使用实战ThreadLocal使用场景我们现在项目中日志记录用的是ELK+Logstash，最后在Kibana中进行展示和检索。现在都是分布式系统统一对外提供服务，项目间调用的关系可以通过 traceId 来关联，但是不同项目之间如何传递 traceId 呢？这里我们使用 org.slf4j.MDC 来实现此功能，内部就是通过 ThreadLocal 来实现的，具体实现如下：当前端发送请求到服务 A时，服务 A会生成一个类似UUID的traceId字符串，将此字符串放入当前线程的ThreadLocal中，在调用服务 B的时候，将traceId写入到请求的Header中，服务 B在接收请求时会先判断请求的Header中是否有traceId，如果存在则写入自己线程的ThreadLocal中。图中的requestId即为我们各个系统链路关联的traceId，系统间互相调用，通过这个requestId即可找到对应链路，这里还有会有一些其他场景：针对于这些场景，我们都可以有相应的解决方案，如下所示Feign 远程调用解决方案服务发送请求：@Component @Slf4j public class FeignInvokeInterceptor implements RequestInterceptor { @Override public void apply(RequestTemplate template) { String requestId = MDC.get("requestId"); if (StringUtils.isNotBlank(requestId)) { template.header("requestId", requestId); } } }服务接收请求：@Slf4j @Component public class LogInterceptor extends HandlerInterceptorAdapter { @Override public void afterCompletion(HttpServletRequest arg0, HttpServletResponse arg1, Object arg2, Exception arg3) { MDC.remove("requestId"); } @Override public void postHandle(HttpServletRequest arg0, HttpServletResponse arg1, Object arg2, ModelAndView arg3) { } @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { String requestId = request.getHeader(BaseConstant.REQUEST_ID_KEY); if (StringUtils.isBlank(requestId)) { requestId = UUID.randomUUID().toString().replace("-", ""); } MDC.put("requestId", requestId); return true; } }线程池异步调用，requestId 传递因为MDC是基于ThreadLocal去实现的，异步过程中，子线程并没有办法获取到父线程ThreadLocal存储的数据，所以这里可以自定义线程池执行器，修改其中的run()方法：public class MyThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { @Override public void execute(Runnable runnable) { Map<String, String> context = MDC.getCopyOfContextMap(); super.execute(() -> run(runnable, context)); } @Override private void run(Runnable runnable, Map<String, String> context) { if (context != null) { MDC.setContextMap(context); } try { runnable.run(); } finally { MDC.remove(); } } }使用 MQ 发送消息给第三方系统在 MQ 发送的消息体中自定义属性requestId，接收方消费消息后，自己解析requestId使用即可。
AI 摘要 文章系统梳理Java线程池：从池化价值、Executor框架、ThreadPoolExecutor七大参数与饱和策略，到execute流程与addWorker源码，对比Runnable/Callable、execute/submit、shutdown/shutdownNow，剖析Fixed、Single、Cached、Scheduled四种内置池的缺陷，给出ThreadPoolExecutor手动创建及阻塞队列选型最佳实践。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: Java 线程池详解category: Javatag:Java并发池化技术想必大家已经屡见不鲜了，线程池、数据库连接池、HTTP 连接池等等都是对这个思想的应用。池化技术的思想主要是为了减少每次获取资源的消耗，提高对资源的利用率。这篇文章我会详细介绍一下线程池的基本概念以及核心原理。线程池介绍顾名思义，线程池就是管理一系列线程的资源池，其提供了一种限制和管理线程资源的方式。每个线程池还维护一些基本统计信息，例如已完成任务的数量。这里借用《Java 并发编程的艺术》书中的部分内容来总结一下使用线程池的好处：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。线程池一般用于执行多个不相关联的耗时任务，没有多线程的情况下，任务顺序执行，使用了线程池的话可让多个不相关联的任务同时执行。Executor 框架介绍Executor 框架是 Java5 之后引进的，在 Java 5 之后，通过 Executor 来启动线程比使用 Thread 的 start 方法更好，除了更易管理，效率更好（用线程池实现，节约开销）外，还有关键的一点：有助于避免 this 逃逸问题。this 逃逸是指在构造函数返回之前其他线程就持有该对象的引用，调用尚未构造完全的对象的方法可能引发令人疑惑的错误。Executor 框架不仅包括了线程池的管理，还提供了线程工厂、队列以及拒绝策略等，Executor 框架让并发编程变得更加简单。Executor 框架结构主要由三大部分组成：1、任务(Runnable /Callable)执行任务需要实现的 Runnable 接口 或 Callable接口。Runnable 接口或 Callable 接口 实现类都可以被 ThreadPoolExecutor 或 ScheduledThreadPoolExecutor 执行。2、任务的执行(Executor)如下图所示，包括任务执行机制的核心接口 Executor ，以及继承自 Executor 接口的 ExecutorService 接口。ThreadPoolExecutor 和 ScheduledThreadPoolExecutor 这两个关键类实现了 ExecutorService 接口。这里提了很多底层的类关系，但是，实际上我们需要更多关注的是 ThreadPoolExecutor 这个类，这个类在我们实际使用线程池的过程中，使用频率还是非常高的。注意： 通过查看 ScheduledThreadPoolExecutor 源代码我们发现 ScheduledThreadPoolExecutor 实际上是继承了 ThreadPoolExecutor 并实现了 ScheduledExecutorService ，而 ScheduledExecutorService 又实现了 ExecutorService，正如我们上面给出的类关系图显示的一样。ThreadPoolExecutor 类描述://AbstractExecutorService实现了ExecutorService接口 public class ThreadPoolExecutor extends AbstractExecutorServiceScheduledThreadPoolExecutor 类描述://ScheduledExecutorService继承ExecutorService接口 public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService3、异步计算的结果(Future)Future 接口以及 Future 接口的实现类 FutureTask 类都可以代表异步计算的结果。当我们把 Runnable接口 或 Callable 接口 的实现类提交给 ThreadPoolExecutor 或 ScheduledThreadPoolExecutor 执行。（调用 submit() 方法时会返回一个 FutureTask 对象）Executor 框架的使用示意图：主线程首先要创建实现 Runnable 或者 Callable 接口的任务对象。把创建完成的实现 Runnable/Callable接口的 对象直接交给 ExecutorService 执行: ExecutorService.execute（Runnable command））或者也可以把 Runnable 对象或Callable 对象提交给 ExecutorService 执行（ExecutorService.submit（Runnable task）或 ExecutorService.submit（Callable <T> task））。如果执行 ExecutorService.submit（…），ExecutorService 将返回一个实现Future接口的对象（我们刚刚也提到过了执行 execute()方法和 submit()方法的区别，submit()会返回一个 FutureTask 对象）。由于 FutureTask 实现了 Runnable，我们也可以创建 FutureTask，然后直接交给 ExecutorService 执行。最后，主线程可以执行 FutureTask.get()方法来等待任务执行完成。主线程也可以执行 FutureTask.cancel（boolean mayInterruptIfRunning）来取消此任务的执行。ThreadPoolExecutor 类介绍（重要）线程池实现类 ThreadPoolExecutor 是 Executor 框架最核心的类。线程池参数分析ThreadPoolExecutor 类中提供的四个构造方法。我们来看最长的那个，其余三个都是在这个构造方法的基础上产生（其他几个构造方法说白点都是给定某些默认参数的构造方法比如默认制定拒绝策略是什么）。 /** * 用给定的初始参数创建一个新的ThreadPoolExecutor。 */ public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量 int maximumPoolSize,//线程池的最大线程数 long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间 TimeUnit unit,//时间单位 BlockingQueue<Runnable> workQueue,//任务队列，用来储存等待执行任务的队列 ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可 RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务 ) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }下面这些参数非常重要，在后面使用线程池的过程中你一定会用到！所以，务必拿着小本本记清楚。ThreadPoolExecutor 3 个最重要的参数：corePoolSize : 任务队列未达到队列容量时，最大可以同时运行的线程数量。maximumPoolSize : 任务队列中存放的任务达到队列容量的时候，当前可以同时运行的线程数量变为最大线程数。workQueue: 新任务来的时候会先判断当前运行的线程数量是否达到核心线程数，如果达到的话，新任务就会被存放在队列中。ThreadPoolExecutor其他常见参数 :keepAliveTime:线程池中的线程数量大于 corePoolSize 的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了 keepAliveTime才会被回收销毁。unit : keepAliveTime 参数的时间单位。threadFactory :executor 创建新线程的时候会用到。handler :饱和策略（后面会单独详细介绍一下）。下面这张图可以加深你对线程池中各个参数的相互关系的理解（图片来源：《Java 性能调优实战》）：ThreadPoolExecutor 饱和策略定义:如果当前同时运行的线程数量达到最大线程数量并且队列也已经被放满了任务时，ThreadPoolExecutor 定义一些策略:ThreadPoolExecutor.AbortPolicy：抛出 RejectedExecutionException来拒绝新任务的处理。ThreadPoolExecutor.CallerRunsPolicy：调用执行自己的线程运行任务，也就是直接在调用execute方法的线程中运行(run)被拒绝的任务，如果执行程序已关闭，则会丢弃该任务。因此这种策略会降低对于新任务提交速度，影响程序的整体性能。如果您的应用程序可以承受此延迟并且你要求任何一个任务请求都要被执行的话，你可以选择这个策略。ThreadPoolExecutor.DiscardPolicy：不处理新任务，直接丢弃掉。ThreadPoolExecutor.DiscardOldestPolicy：此策略将丢弃最早的未处理的任务请求。举个例子：举个例子：Spring 通过 ThreadPoolTaskExecutor 或者我们直接通过 ThreadPoolExecutor 的构造函数创建线程池的时候，当我们不指定 RejectedExecutionHandler 饱和策略来配置线程池的时候，默认使用的是 AbortPolicy。在这种饱和策略下，如果队列满了，ThreadPoolExecutor 将抛出 RejectedExecutionException 异常来拒绝新来的任务 ，这代表你将丢失对这个任务的处理。如果不想丢弃任务的话，可以使用CallerRunsPolicy。CallerRunsPolicy 和其他的几个策略不同，它既不会抛弃任务，也不会抛出异常，而是将任务回退给调用者，使用调用者的线程来执行任务public static class CallerRunsPolicy implements RejectedExecutionHandler { public CallerRunsPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { // 直接主线程执行，而不是线程池中的线程执行 r.run(); } } }线程池创建的两种方式方式一：通过ThreadPoolExecutor构造函数来创建（推荐）。方式二：通过 Executor 框架的工具类 Executors 来创建。Executors工具类提供的创建线程池的方法如下图所示：可以看出，通过Executors工具类可以创建多种类型的线程池，包括：FixedThreadPool：固定线程数量的线程池。该线程池中的线程数量始终不变。当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。若没有，则新的任务会被暂存在一个任务队列中，待有线程空闲时，便处理在任务队列中的任务。SingleThreadExecutor： 只有一个线程的线程池。若多余一个任务被提交到该线程池，任务会被保存在一个任务队列中，待线程空闲，按先入先出的顺序执行队列中的任务。CachedThreadPool： 可根据实际情况调整线程数量的线程池。线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。若所有线程均在工作，又有新的任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用。ScheduledThreadPool：给定的延迟后运行任务或者定期执行任务的线程池。《阿里巴巴 Java 开发手册》强制线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 构造函数的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险Executors 返回线程池对象的弊端如下(后文会详细介绍到)：FixedThreadPool 和 SingleThreadExecutor:使用的是无界的 LinkedBlockingQueue，任务队列最大长度为 Integer.MAX_VALUE,可能堆积大量的请求，从而导致 OOM。CachedThreadPool:使用的是同步队列 SynchronousQueue, 允许创建的线程数量为 Integer.MAX_VALUE ，如果任务数量过多且执行速度较慢，可能会创建大量的线程，从而导致 OOM。ScheduledThreadPool 和 SingleThreadScheduledExecutor:使用的无界的延迟阻塞队列DelayedWorkQueue，任务队列最大长度为 Integer.MAX_VALUE,可能堆积大量的请求，从而导致 OOM。// 无界队列 LinkedBlockingQueue public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()); } // 无界队列 LinkedBlockingQueue public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>())); } // 同步队列 SynchronousQueue，没有容量，最大线程数是 Integer.MAX_VALUE` public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>()); } // DelayedWorkQueue（延迟阻塞队列） public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }线程池常用的阻塞队列总结新任务来的时候会先判断当前运行的线程数量是否达到核心线程数，如果达到的话，新任务就会被存放在队列中。不同的线程池会选用不同的阻塞队列，我们可以结合内置线程池来分析。容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列）：FixedThreadPool 和 SingleThreadExector 。FixedThreadPool最多只能创建核心线程数的线程（核心线程数和最大线程数相等），SingleThreadExector只能创建一个线程（核心线程数和最大线程数都是 1），二者的任务队列永远不会被放满。SynchronousQueue（同步队列）：CachedThreadPool 。SynchronousQueue 没有容量，不存储元素，目的是保证对于提交的任务，如果有空闲线程，则使用空闲线程来处理；否则新建一个线程来处理任务。也就是说，CachedThreadPool 的最大线程数是 Integer.MAX_VALUE ，可以理解为线程数是可以无限扩展的，可能会创建大量线程，从而导致 OOM。DelayedWorkQueue（延迟阻塞队列）：ScheduledThreadPool 和 SingleThreadScheduledExecutor 。DelayedWorkQueue 的内部元素并不是按照放入的时间排序，而是会按照延迟的时间长短对任务进行排序，内部采用的是“堆”的数据结构，可以保证每次出队的任务都是当前队列中执行时间最靠前的。DelayedWorkQueue 添加元素满了之后会自动扩容原来容量的 1/2，即永远不会阻塞，最大扩容可达 Integer.MAX_VALUE，所以最多只能创建核心线程数的线程。线程池原理分析（重要）我们上面讲解了 Executor框架以及 ThreadPoolExecutor 类，下面让我们实战一下，来通过写一个 ThreadPoolExecutor 的小 Demo 来回顾上面的内容。线程池示例代码首先创建一个 Runnable 接口的实现类（当然也可以是 Callable 接口，我们后面会介绍两者的区别。）MyRunnable.javaimport java.util.Date; /** * 这是一个简单的Runnable类，需要大约5秒钟来执行其任务。 * @author shuang.kou */ public class MyRunnable implements Runnable { private String command; public MyRunnable(String s) { this.command = s; } @Override public void run() { System.out.println(Thread.currentThread().getName() + " Start. Time = " + new Date()); processCommand(); System.out.println(Thread.currentThread().getName() + " End. Time = " + new Date()); } private void processCommand() { try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } } @Override public String toString() { return this.command; } } 编写测试程序，我们这里以阿里巴巴推荐的使用 ThreadPoolExecutor 构造函数自定义参数的方式来创建线程池。ThreadPoolExecutorDemo.javaimport java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class ThreadPoolExecutorDemo { private static final int CORE_POOL_SIZE = 5; private static final int MAX_POOL_SIZE = 10; private static final int QUEUE_CAPACITY = 100; private static final Long KEEP_ALIVE_TIME = 1L; public static void main(String[] args) { //使用阿里巴巴推荐的创建线程池的方式 //通过ThreadPoolExecutor构造函数自定义参数创建 ThreadPoolExecutor executor = new ThreadPoolExecutor( CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue<>(QUEUE_CAPACITY), new ThreadPoolExecutor.CallerRunsPolicy()); for (int i = 0; i < 10; i++) { //创建WorkerThread对象（WorkerThread类实现了Runnable 接口） Runnable worker = new MyRunnable("" + i); //执行Runnable executor.execute(worker); } //终止线程池 executor.shutdown(); while (!executor.isTerminated()) { } System.out.println("Finished all threads"); } } 可以看到我们上面的代码指定了：corePoolSize: 核心线程数为 5。maximumPoolSize：最大线程数 10keepAliveTime : 等待时间为 1L。unit: 等待时间的单位为 TimeUnit.SECONDS。workQueue：任务队列为 ArrayBlockingQueue，并且容量为 100;handler:饱和策略为 CallerRunsPolicy。输出结构：pool-1-thread-3 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-5 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-2 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-1 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-4 Start. Time = Sun Apr 12 11:14:37 CST 2020 pool-1-thread-3 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-4 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-1 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-5 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-1 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-2 End. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-5 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-4 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-3 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-2 Start. Time = Sun Apr 12 11:14:42 CST 2020 pool-1-thread-1 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-4 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-5 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-3 End. Time = Sun Apr 12 11:14:47 CST 2020 pool-1-thread-2 End. Time = Sun Apr 12 11:14:47 CST 2020 Finished all threads // 任务全部执行完了才会跳出来，因为executor.isTerminated()判断为true了才会跳出while循环，当且仅当调用 shutdown() 方法后，并且所有提交的任务完成后返回为 true 线程池原理分析我们通过前面的代码输出结果可以看出：线程池首先会先执行 5 个任务，然后这些任务有任务被执行完的话，就会去拿新的任务执行。 大家可以先通过上面讲解的内容，分析一下到底是咋回事？（自己独立思考一会）现在，我们就分析上面的输出内容来简单分析一下线程池原理。为了搞懂线程池的原理，我们需要首先分析一下 execute方法。 在示例代码中，我们使用 executor.execute(worker)来提交一个任务到线程池中去。这个方法非常重要，下面我们来看看它的源码： // 存放线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount) private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static int workerCountOf(int c) { return c & CAPACITY; } //任务队列 private final BlockingQueue<Runnable> workQueue; public void execute(Runnable command) { // 如果任务为null，则抛出异常。 if (command == null) throw new NullPointerException(); // ctl 中保存的线程池当前的一些状态信息 int c = ctl.get(); // 下面会涉及到 3 步 操作 // 1.首先判断当前线程池中执行的任务数量是否小于 corePoolSize // 如果小于的话，通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 2.如果当前执行的任务数量大于等于 corePoolSize 的时候就会走到这里，表明创建新的线程失败。 // 通过 isRunning 方法判断线程池状态，线程池处于 RUNNING 状态并且队列可以加入任务，该任务才会被加入进去 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 再次获取线程池状态，如果线程池状态不是 RUNNING 状态就需要从任务队列中移除任务，并尝试判断线程是否全部执行完毕。同时执行拒绝策略。 if (!isRunning(recheck) && remove(command)) reject(command); // 如果当前工作线程数量为0，新创建一个线程并执行。 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //3. 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。 // 传入 false 代表增加线程时判断当前线程数是否少于 maxPoolSize //如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。 else if (!addWorker(command, false)) reject(command); }这里简单分析一下整个流程（对整个逻辑进行了简化，方便理解）：如果当前运行的线程数小于核心线程数，那么就会新建一个线程来执行任务。如果当前运行的线程数等于或大于核心线程数，但是小于最大线程数，那么就把该任务放入到任务队列里等待执行。如果向任务队列投放任务失败（任务队列已经满了），但是当前运行的线程数是小于最大线程数的，就新建一个线程来执行任务。如果当前运行的线程数已经等同于最大线程数了，新建线程将会使当前运行的线程超出最大线程数，那么当前任务会被拒绝，饱和策略会调用RejectedExecutionHandler.rejectedExecution()方法。在 execute 方法中，多次调用 addWorker 方法。addWorker 这个方法主要用来创建新的工作线程，如果返回 true 说明创建和启动工作线程成功，否则的话返回的就是 false。 // 全局锁，并发操作必备 private final ReentrantLock mainLock = new ReentrantLock(); // 跟踪线程池的最大大小，只有在持有全局锁mainLock的前提下才能访问此集合 private int largestPoolSize; // 工作线程集合，存放线程池中所有的（活跃的）工作线程，只有在持有全局锁mainLock的前提下才能访问此集合 private final HashSet<Worker> workers = new HashSet<>(); //获取线程池状态 private static int runStateOf(int c) { return c & ~CAPACITY; } //判断线程池的状态是否为 Running private static boolean isRunning(int c) { return c < SHUTDOWN; } /** * 添加新的工作线程到线程池 * @param firstTask 要执行 * @param core参数为true的话表示使用线程池的基本大小，为false使用线程池最大大小 * @return 添加成功就返回true否则返回false */ private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { //这两句用来获取线程池的状态 int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { //获取线程池中工作的线程的数量 int wc = workerCountOf(c); // core参数为false的话表明队列也满了，线程池大小变为 maximumPoolSize if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //原子操作将workcount的数量加1 if (compareAndIncrementWorkerCount(c)) break retry; // 如果线程的状态改变了就再次执行上述操作 c = ctl.get(); if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } // 标记工作线程是否启动成功 boolean workerStarted = false; // 标记工作线程是否创建成功 boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // 加锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //获取线程池状态 int rs = runStateOf(ctl.get()); //rs < SHUTDOWN 如果线程池状态依然为RUNNING,并且线程的状态是存活的话，就会将工作线程添加到工作线程集合中 //(rs=SHUTDOWN && firstTask == null)如果线程池状态小于STOP，也就是RUNNING或者SHUTDOWN状态下，同时传入的任务实例firstTask为null，则需要添加到工作线程集合和启动新的Worker // firstTask == null证明只新建线程而不执行任务 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); //更新当前工作线程的最大容量 int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; // 工作线程是否启动成功 workerAdded = true; } } finally { // 释放锁 mainLock.unlock(); } //// 如果成功添加工作线程，则调用Worker内部的线程实例t的Thread#start()方法启动真实的线程实例 if (workerAdded) { t.start(); /// 标记线程启动成功 workerStarted = true; } } } finally { // 线程启动失败，需要从工作线程中移除对应的Worker if (! workerStarted) addWorkerFailed(w); } return workerStarted; }更多关于线程池源码分析的内容推荐这篇文章：硬核干货：4W 字从源码上分析 JUC 线程池 ThreadPoolExecutor 的实现原理现在，让我们在回到示例代码， 现在应该是不是很容易就可以搞懂它的原理了呢？没搞懂的话，也没关系，可以看看我的分析：我们在代码中模拟了 10 个任务，我们配置的核心线程数为 5、等待队列容量为 100 ，所以每次只可能存在 5 个任务同时执行，剩下的 5 个任务会被放到等待队列中去。当前的 5 个任务中如果有任务被执行完了，线程池就会去拿新的任务执行。几个常见的对比Runnable vs CallableRunnable自 Java 1.0 以来一直存在，但Callable仅在 Java 1.5 中引入,目的就是为了来处理Runnable不支持的用例。Runnable 接口不会返回结果或抛出检查异常，但是 Callable 接口可以。所以，如果任务不需要返回结果或抛出异常推荐使用 Runnable 接口，这样代码看起来会更加简洁。工具类 Executors 可以实现将 Runnable 对象转换成 Callable 对象。（Executors.callable(Runnable task) 或 Executors.callable(Runnable task, Object result)）。Runnable.java@FunctionalInterface public interface Runnable { /** * 被线程执行，没有返回值也无法抛出异常 */ public abstract void run(); }Callable.java@FunctionalInterface public interface Callable<V> { /** * 计算结果，或在无法这样做时抛出异常。 * @return 计算得出的结果 * @throws 如果无法计算结果，则抛出异常 */ V call() throws Exception; } execute() vs submit()execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否；submit()方法用于提交需要返回值的任务。线程池会返回一个 Future 类型的对象，通过这个 Future 对象可以判断任务是否执行成功，并且可以通过 Future 的 get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用 get（long timeout，TimeUnit unit）方法的话，如果在 timeout 时间内任务还没有执行完，就会抛出 java.util.concurrent.TimeoutException。这里只是为了演示使用，推荐使用 ThreadPoolExecutor 构造方法来创建线程池。示例 1：使用 get()方法获取返回值。ExecutorService executorService = Executors.newFixedThreadPool(3); Future<String> submit = executorService.submit(() -> { try { Thread.sleep(5000L); } catch (InterruptedException e) { e.printStackTrace(); } return "abc"; }); String s = submit.get(); System.out.println(s); executorService.shutdown();输出：abc示例 2：使用 get（long timeout，TimeUnit unit）方法获取返回值。ExecutorService executorService = Executors.newFixedThreadPool(3); Future<String> submit = executorService.submit(() -> { try { Thread.sleep(5000L); } catch (InterruptedException e) { e.printStackTrace(); } return "abc"; }); String s = submit.get(3, TimeUnit.SECONDS); System.out.println(s); executorService.shutdown();输出：Exception in thread "main" java.util.concurrent.TimeoutException at java.util.concurrent.FutureTask.get(FutureTask.java:205)shutdown()VSshutdownNow()shutdown（） :关闭线程池，线程池的状态变为 SHUTDOWN。线程池不再接受新任务了，但是队列里的任务得执行完毕。shutdownNow（） :关闭线程池，线程池的状态变为 STOP。线程池会终止当前正在运行的任务，并停止处理排队的任务并返回正在等待执行的 List。isTerminated() VS isShutdown()isShutDown 当调用 shutdown() 方法后返回为 true。isTerminated 当调用 shutdown() 方法后，并且所有提交的任务完成后返回为 true几种常见的内置线程池FixedThreadPool介绍FixedThreadPool 被称为可重用固定线程数的线程池。通过 Executors 类中的相关源代码来看一下相关实现： /** * 创建一个可重用固定数量线程的线程池 */ public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory); }另外还有一个 FixedThreadPool 的实现方法，和上面的类似，所以这里不多做阐述： public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }从上面源代码可以看出新创建的 FixedThreadPool 的 corePoolSize 和 maximumPoolSize 都被设置为 nThreads，这个 nThreads 参数是我们使用的时候自己传递的。即使 maximumPoolSize 的值比 corePoolSize 大，也至多只会创建 corePoolSize 个线程。这是因为FixedThreadPool 使用的是容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列），队列永远不会被放满。执行任务过程介绍FixedThreadPool 的 execute() 方法运行示意图（该图片来源：《Java 并发编程的艺术》）：上图说明：如果当前运行的线程数小于 corePoolSize， 如果再来新任务的话，就创建新的线程来执行任务；当前运行的线程数等于 corePoolSize 后， 如果再来新任务的话，会将任务加入 LinkedBlockingQueue；线程池中的线程执行完 手头的任务后，会在循环中反复从 LinkedBlockingQueue 中获取任务来执行；为什么不推荐使用FixedThreadPool？FixedThreadPool 使用无界队列 LinkedBlockingQueue（队列的容量为 Integer.MAX_VALUE）作为线程池的工作队列会对线程池带来如下影响：当线程池中的线程数达到 corePoolSize 后，新任务将在无界队列中等待，因此线程池中的线程数不会超过 corePoolSize；由于使用无界队列时 maximumPoolSize 将是一个无效参数，因为不可能存在任务队列满的情况。所以，通过创建 FixedThreadPool的源码可以看出创建的 FixedThreadPool 的 corePoolSize 和 maximumPoolSize 被设置为同一个值。由于 1 和 2，使用无界队列时 keepAliveTime 将是一个无效参数；运行中的 FixedThreadPool（未执行 shutdown()或 shutdownNow()）不会拒绝任务，在任务比较多的时候会导致 OOM（内存溢出）。SingleThreadExecutor介绍SingleThreadExecutor 是只有一个线程的线程池。下面看看SingleThreadExecutor 的实现： /** *返回只有一个线程的线程池 */ public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory)); } public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }从上面源代码可以看出新创建的 SingleThreadExecutor 的 corePoolSize 和 maximumPoolSize 都被设置为 1，其他参数和 FixedThreadPool 相同。执行任务过程介绍SingleThreadExecutor 的运行示意图（该图片来源：《Java 并发编程的艺术》）：上图说明 :如果当前运行的线程数少于 corePoolSize，则创建一个新的线程执行任务；当前线程池中有一个运行的线程后，将任务加入 LinkedBlockingQueue线程执行完当前的任务后，会在循环中反复从LinkedBlockingQueue 中获取任务来执行；为什么不推荐使用SingleThreadExecutor？SingleThreadExecutor 和 FixedThreadPool 一样，使用的都是容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue（无界队列）作为线程池的工作队列。SingleThreadExecutor 使用无界队列作为线程池的工作队列会对线程池带来的影响与 FixedThreadPool 相同。说简单点，就是可能会导致 OOM。CachedThreadPool介绍CachedThreadPool 是一个会根据需要创建新线程的线程池。下面通过源码来看看 CachedThreadPool 的实现： /** * 创建一个线程池，根据需要创建新线程，但会在先前构建的线程可用时重用它。 */ public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), threadFactory); } public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }CachedThreadPool 的corePoolSize 被设置为空（0），maximumPoolSize被设置为 Integer.MAX.VALUE，即它是无界的，这也就意味着如果主线程提交任务的速度高于 maximumPool 中线程处理任务的速度时，CachedThreadPool 会不断创建新的线程。极端情况下，这样会导致耗尽 cpu 和内存资源。执行任务过程介绍CachedThreadPool 的 execute() 方法的执行示意图（该图片来源：《Java 并发编程的艺术》）：上图说明：首先执行 SynchronousQueue.offer(Runnable task) 提交任务到任务队列。如果当前 maximumPool 中有闲线程正在执行 SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)，那么主线程执行 offer 操作与空闲线程执行的 poll 操作配对成功，主线程把任务交给空闲线程执行，execute()方法执行完成，否则执行下面的步骤 2；当初始 maximumPool 为空，或者 maximumPool 中没有空闲线程时，将没有线程执行 SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)。这种情况下，步骤 1 将失败，此时 CachedThreadPool 会创建新线程执行任务，execute 方法执行完成；为什么不推荐使用CachedThreadPool？CachedThreadPool 使用的是同步队列 SynchronousQueue, 允许创建的线程数量为 Integer.MAX_VALUE ，可能会创建大量线程，从而导致 OOM。ScheduledThreadPool介绍ScheduledThreadPool 用来在给定的延迟后运行任务或者定期执行任务。这个在实际项目中基本不会被用到，也不推荐使用，大家只需要简单了解一下即可。public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }ScheduledThreadPool 是通过 ScheduledThreadPoolExecutor 创建的，使用的DelayedWorkQueue（延迟阻塞队列）作为线程池的任务队列。DelayedWorkQueue 的内部元素并不是按照放入的时间排序，而是会按照延迟的时间长短对任务进行排序，内部采用的是“堆”的数据结构，可以保证每次出队的任务都是当前队列中执行时间最靠前的。DelayedWorkQueue 添加元素满了之后会自动扩容原来容量的 1/2，即永远不会阻塞，最大扩容可达 Integer.MAX_VALUE，所以最多只能创建核心线程数的线程。ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor，所以创建 ScheduledThreadExecutor 本质也是创建一个 ThreadPoolExecutor 线程池，只是传入的参数不相同。public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorServiceScheduledThreadPoolExecutor 和 Timer 对比Timer 对系统时钟的变化敏感，ScheduledThreadPoolExecutor不是；Timer 只有一个执行线程，因此长时间运行的任务可以延迟其他任务。 ScheduledThreadPoolExecutor 可以配置任意数量的线程。 此外，如果你想（通过提供 ThreadFactory），你可以完全控制创建的线程;在TimerTask 中抛出的运行时异常会杀死一个线程，从而导致 Timer 死机即计划任务将不再运行。ScheduledThreadExecutor 不仅捕获运行时异常，还允许您在需要时处理它们（通过重写 afterExecute 方法ThreadPoolExecutor）。抛出异常的任务将被取消，但其他任务将继续运行。关于定时任务的详细介绍，可以看这篇文章：Java 定时任务详解 。线程池最佳实践Java 线程池最佳实践这篇文章总结了一些使用线程池的时候应该注意的东西，实际项目使用线程池之前可以看看。
AI 摘要 文章系统梳理了ArrayBlockingQueue：从JDK1.5阻塞队列历史、生产者-消费者思想，到循环数组+ReentrantLock+双Condition的源码实现，对比put/offer/add、take/poll/peek等方法差异，并给出与LinkedBlockingQueue、ConcurrentLinkedQueue的区别及面试要点。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: ArrayBlockingQueue 源码分析category: Javatag:Java集合阻塞队列简介阻塞队列的历史Java 阻塞队列的历史可以追溯到 JDK1.5 版本，当时 Java 平台增加了 java.util.concurrent，即我们常说的 JUC 包，其中包含了各种并发流程控制工具、并发容器、原子类等。这其中自然也包含了我们这篇文章所讨论的阻塞队列。为了解决高并发场景下多线程之间数据共享的问题，JDK1.5 版本中出现了 ArrayBlockingQueue 和 LinkedBlockingQueue，它们是带有生产者-消费者模式实现的并发容器。其中，ArrayBlockingQueue 是有界队列，即添加的元素达到上限之后，再次添加就会被阻塞或者抛出异常。而 LinkedBlockingQueue 则由链表构成的队列，正是因为链表的特性，所以 LinkedBlockingQueue 在添加元素上并不会向 ArrayBlockingQueue 那样有着较多的约束，所以 LinkedBlockingQueue 设置队列是否有界是可选的(注意这里的无界并不是指可以添加任务数量的元素，而是说队列的大小默认为 Integer.MAX_VALUE，近乎于无限大)。随着 Java 的不断发展，JDK 后续的几个版本又对阻塞队列进行了不少的更新和完善:JDK1.6 版本:增加 SynchronousQueue，一个不存储元素的阻塞队列。JDK1.7 版本:增加 TransferQueue，一个支持更多操作的阻塞队列。JDK1.8 版本:增加 DelayQueue，一个支持延迟获取元素的阻塞队列。阻塞队列的思想阻塞队列就是典型的生产者-消费者模型，它可以做到以下几点:当阻塞队列数据为空时，所有的消费者线程都会被阻塞，等待队列非空。当生产者往队列里填充数据后，队列就会通知消费者队列非空，消费者此时就可以进来消费。当阻塞队列因为消费者消费过慢或者生产者存放元素过快导致队列填满时无法容纳新元素时，生产者就会被阻塞，等待队列非满时继续存放元素。当消费者从队列中消费一个元素之后，队列就会通知生产者队列非满，生产者可以继续填充数据了。总结一下：阻塞队列就说基于非空和非满两个条件实现生产者和消费者之间的交互，尽管这些交互流程和等待通知的机制实现非常复杂，好在 Doug Lea 的操刀之下已将阻塞队列的细节屏蔽，我们只需调用 put、take、offer、poll 等 API 即可实现多线程之间的生产和消费。这也使得阻塞队列在多线程开发中有着广泛的运用，最常见的例子无非是我们的线程池,从源码中我们就能看出当核心线程无法及时处理任务时，这些任务都会扔到 workQueue 中。public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {// ...}ArrayBlockingQueue 常见方法及测试简单了解了阻塞队列的历史之后，我们就开始重点讨论本篇文章所要介绍的并发容器——ArrayBlockingQueue。为了后续更加深入的了解 ArrayBlockingQueue，我们不妨基于下面几个实例了解以下 ArrayBlockingQueue 的使用。先看看第一个例子，我们这里会用两个线程分别模拟生产者和消费者，生产者生产完会使用 put 方法生产 10 个元素给消费者进行消费，当队列元素达到我们设置的上限 5 时，put 方法就会阻塞。同理消费者也会通过 take 方法消费元素，当队列为空时，take 方法就会阻塞消费者线程。这里笔者为了保证消费者能够在消费完 10 个元素后及时退出。便通过倒计时门闩，来控制消费者结束，生产者在这里只会生产 10 个元素。当消费者将 10 个元素消费完成之后，按下倒计时门闩，所有线程都会停止。public class ProducerConsumerExample { public static void main(String[] args) throws InterruptedException { // 创建一个大小为 5 的 ArrayBlockingQueue ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5); // 创建生产者线程 Thread producer = new Thread(() -> { try { for (int i = 1; i <= 10; i++) { // 向队列中添加元素，如果队列已满则阻塞等待 queue.put(i); System.out.println("生产者添加元素：" + i); } } catch (InterruptedException e) { e.printStackTrace(); } }); CountDownLatch countDownLatch = new CountDownLatch(1); // 创建消费者线程 Thread consumer = new Thread(() -> { try { int count = 0; while (true) { // 从队列中取出元素，如果队列为空则阻塞等待 int element = queue.take(); System.out.println("消费者取出元素：" + element); ++count; if (count == 10) { break; } } countDownLatch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } }); // 启动线程 producer.start(); consumer.start(); // 等待线程结束 producer.join(); consumer.join(); countDownLatch.await(); producer.interrupt(); consumer.interrupt(); } }代码输出结果如下，可以看到只有生产者往队列中投放元素之后消费者才能消费，这也就意味着当队列中没有数据的时消费者就会阻塞，等待队列非空再继续消费。生产者添加元素：1 生产者添加元素：2 消费者取出元素：1 消费者取出元素：2 消费者取出元素：3 生产者添加元素：3 生产者添加元素：4 生产者添加元素：5 消费者取出元素：4 生产者添加元素：6 消费者取出元素：5 生产者添加元素：7 生产者添加元素：8 生产者添加元素：9 生产者添加元素：10 消费者取出元素：6 消费者取出元素：7 消费者取出元素：8 消费者取出元素：9 消费者取出元素：10了解了 put、take 这两个会阻塞的存和取方法之后，我我们再来看看阻塞队列中非阻塞的入队和出队方法 offer 和 poll。如下所示，我们设置了一个大小为 3 的阻塞队列，我们会尝试在队列用 offer 方法存放 4 个元素，然后再从队列中用 poll 尝试取 4 次。public class OfferPollExample { public static void main(String[] args) { // 创建一个大小为 3 的 ArrayBlockingQueue ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(3); // 向队列中添加元素 System.out.println(queue.offer("A")); System.out.println(queue.offer("B")); System.out.println(queue.offer("C")); // 尝试向队列中添加元素，但队列已满，返回 false System.out.println(queue.offer("D")); // 从队列中取出元素 System.out.println(queue.poll()); System.out.println(queue.poll()); System.out.println(queue.poll()); // 尝试从队列中取出元素，但队列已空，返回 null System.out.println(queue.poll()); } }最终代码的输出结果如下，可以看到因为队列的大小为 3 的缘故，我们前 3 次存放到队列的结果为 true，第 4 次存放时，由于队列已满，所以存放结果返回 false。这也是为什么我们后续的 poll 方法只得到了 3 个元素的值。true true true false A B C null了解了阻塞存取和非阻塞存取，我们再来看看阻塞队列的一个比较特殊的操作，某些场景下，我们希望能够一次性将阻塞队列的结果存到列表中再进行批量操作，我们就可以使用阻塞队列的 drainTo 方法，这个方法会一次性将队列中所有元素存放到列表，如果队列中有元素，且成功存到 list 中则 drainTo 会返回本次转移到 list 中的元素数，反之若队列为空，drainTo 则直接返回 0。public class DrainToExample { public static void main(String[] args) { // 创建一个大小为 5 的 ArrayBlockingQueue ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5); // 向队列中添加元素 queue.add(1); queue.add(2); queue.add(3); queue.add(4); queue.add(5); // 创建一个 List，用于存储从队列中取出的元素 List<Integer> list = new ArrayList<>(); // 从队列中取出所有元素，并添加到 List 中 queue.drainTo(list); // 输出 List 中的元素 System.out.println(list); } }代码输出结果如下[1, 2, 3, 4, 5]ArrayBlockingQueue 源码分析自此我们对阻塞队列的使用有了基本的印象，接下来我们就可以进一步了解一下 ArrayBlockingQueue 的工作机制了。整体设计在了解 ArrayBlockingQueue 的具体细节之前，我们先来看看 ArrayBlockingQueue 的类图。从图中我们可以看出，ArrayBlockingQueue 继承了阻塞队列 BlockingQueue 这个接口，不难猜出通过继承 BlockingQueue 这个接口之后，ArrayBlockingQueue 就拥有了阻塞队列那些常见的操作行为。同时， ArrayBlockingQueue 还继承了 AbstractQueue 这个抽象类，这个继承了 AbstractCollection 和 Queue 的抽象类，从抽象类的特定和语义我们也可以猜出，这个继承关系使得 ArrayBlockingQueue 拥有了队列的常见操作。所以我们是否可以得出这样一个结论，通过继承 AbstractQueue 获得队列所有的操作模板，其实现的入队和出队操作的整体框架。然后 ArrayBlockingQueue 通过继承 BlockingQueue 获取到阻塞队列的常见操作并将这些操作实现，填充到 AbstractQueue 模板方法的细节中，由此 ArrayBlockingQueue 成为一个完整的阻塞队列。为了印证这一点，我们到源码中一探究竟。首先我们先来看看 AbstractQueue，从类的继承关系我们可以大致得出，它通过 AbstractCollection 获得了集合的常见操作方法，然后通过 Queue 接口获得了队列的特性。public abstract class AbstractQueue<E> extends AbstractCollection<E> implements Queue<E> { //... }对于集合的操作无非是增删改查，所以我们不妨从添加方法入手，从源码中我们可以看到，它实现了 AbstractCollection 的 add 方法，其内部逻辑如下:调用继承 Queue 接口的来的 offer 方法，如果 offer 成功则返回 true。如果 offer 失败，即代表当前元素入队失败直接抛异常。public boolean add(E e) { if (offer(e)) return true; else throw new IllegalStateException("Queue full"); }而 AbstractQueue 中并没有对 Queue 的 offer 的实现，很明显这样做的目的是定义好了 add 的核心逻辑，将 offer 的细节交由其子类即我们的 ArrayBlockingQueue 实现。到此，我们对于抽象类 AbstractQueue 的分析就结束了，我们继续看看 ArrayBlockingQueue 中另一个重要的继承接口 BlockingQueue。点开 BlockingQueue 之后，我们可以看到这个接口同样继承了 Queue 接口，这就意味着它也具备了队列所拥有的所有行为。同时，它还定义了自己所需要实现的方法。public interface BlockingQueue<E> extends Queue<E> { //元素入队成功返回true，反之则会抛出异常IllegalStateException boolean add(E e); //元素入队成功返回true，反之返回false boolean offer(E e); //元素入队成功则直接返回，如果队列已满元素不可入队则将线程阻塞，因为阻塞期间可能会被打断，所以这里方法签名抛出了InterruptedException void put(E e) throws InterruptedException; //和上一个方法一样,只不过队列满时只会阻塞单位为unit，时间为timeout的时长，如果在等待时长内没有入队成功则直接返回false。 boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; //从队头取出一个元素，如果队列为空则阻塞等待，因为会阻塞线程的缘故，所以该方法可能会被打断，所以签名定义了InterruptedException E take() throws InterruptedException; //取出队头的元素并返回，如果当前队列为空则阻塞等待timeout且单位为unit的时长，如果这个时间段没有元素则直接返回null。 E poll(long timeout, TimeUnit unit) throws InterruptedException; //获取队列剩余元素个数 int remainingCapacity(); //删除我们指定的对象，如果成功返回true，反之返回false。 boolean remove(Object o); //判断队列中是否包含指定元素 public boolean contains(Object o); //将队列中的元素全部存到指定的集合中 int drainTo(Collection<? super E> c); //转移maxElements个元素到集合中 int drainTo(Collection<? super E> c, int maxElements); }了解了 BlockingQueue 的常见操作后，我们就知道了 ArrayBlockingQueue 通过继承 BlockingQueue 的方法并实现后，填充到 AbstractQueue 的方法上，由此我们便知道了上文中 AbstractQueue 的 add 方法的 offer 方法是哪里是实现的了。public boolean add(E e) { //AbstractQueue的offer来自下层的ArrayBlockingQueue从BlockingQueue继承并实现的offer方法 if (offer(e)) return true; else throw new IllegalStateException("Queue full"); }初始化了解 ArrayBlockingQueue 的细节前，我们不妨先看看其构造函数，了解一下其初始化过程。从源码中我们可以看出 ArrayBlockingQueue 有 3 个构造方法，而最核心的构造方法就是下方这一个。// capacity 表示队列初始容量，fair 表示 锁的公平性 public ArrayBlockingQueue(int capacity, boolean fair) { //如果设置的队列大小小于0，则直接抛出IllegalArgumentException if (capacity <= 0) throw new IllegalArgumentException(); //初始化一个数组用于存放队列的元素 this.items = new Object[capacity]; //创建阻塞队列流程控制的锁 lock = new ReentrantLock(fair); //用lock锁创建两个条件控制队列生产和消费 notEmpty = lock.newCondition(); notFull = lock.newCondition(); }这个构造方法里面有两个比较核心的成员变量 notEmpty(非空) 和 notFull （非满） ，需要我们格外留意，它们是实现生产者和消费者有序工作的关键所在，这一点笔者会在后续的源码解析中详细说明，这里我们只需初步了解一下阻塞队列的构造即可。另外两个构造方法都是基于上述的构造方法，默认情况下，我们会使用下面这个构造方法，该构造方法就意味着 ArrayBlockingQueue 用的是非公平锁，即各个生产者或者消费者线程收到通知后，对于锁的争抢是随机的。 public ArrayBlockingQueue(int capacity) { this(capacity, false); }还有一个不怎么常用的构造方法，在初始化容量和锁的非公平性之后，它还提供了一个 Collection 参数，从源码中不难看出这个构造方法是将外部传入的集合的元素在初始化时直接存放到阻塞队列中。public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) { //初始化容量和锁的公平性 this(capacity, fair); final ReentrantLock lock = this.lock; //上锁并将c中的元素存放到ArrayBlockingQueue底层的数组中 lock.lock(); try { int i = 0; try { //遍历并添加元素到数组中 for (E e : c) { checkNotNull(e); items[i++] = e; } } catch (ArrayIndexOutOfBoundsException ex) { throw new IllegalArgumentException(); } //记录当前队列容量 count = i; //更新下一次put或者offer或用add方法添加到队列底层数组的位置 putIndex = (i == capacity) ? 0 : i; } finally { //完成遍历后释放锁 lock.unlock(); } }阻塞式获取和新增元素ArrayBlockingQueue 阻塞式获取和新增元素对应的就是生产者-消费者模型，虽然它也支持非阻塞式获取和新增元素（例如 poll() 和 offer(E e) 方法，后文会介绍到），但一般不会使用。ArrayBlockingQueue 阻塞式获取和新增元素的方法为：put(E e)：将元素插入队列中，如果队列已满，则该方法会一直阻塞，直到队列有空间可用或者线程被中断。take() ：获取并移除队列头部的元素，如果队列为空，则该方法会一直阻塞，直到队列非空或者线程被中断。这两个方法实现的关键就是在于两个条件对象 notEmpty(非空) 和 notFull （非满），这个我们在上文的构造方法中有提到。接下来笔者就通过两张图让大家了解一下这两个条件是如何在阻塞队列中运用的。假设我们的代码消费者先启动，当它发现队列中没有数据，那么非空条件就会将这个线程挂起，即等待条件非空时挂起。然后 CPU 执行权到达生产者，生产者发现队列中可以存放数据，于是将数据存放进去，通知此时条件非空，此时消费者就会被唤醒到队列中使用 take 等方法获取值了。随后的执行中，生产者生产速度远远大于消费者消费速度，于是生产者将队列塞满后再次尝试将数据存入队列，发现队列已满，于是阻塞队列就将当前线程挂起，等待非满。然后消费者拿着 CPU 执行权进行消费，于是队列可以存放新数据了，发出一个非满的通知，此时挂起的生产者就会等待 CPU 执行权到来时再次尝试将数据存到队列中。简单了解阻塞队列的基于两个条件的交互流程之后，我们不妨看看 put 和 take 方法的源码。public void put(E e) throws InterruptedException { //确保插入的元素不为null checkNotNull(e); //加锁 final ReentrantLock lock = this.lock; //这里使用lockInterruptibly()方法而不是lock()方法是为了能够响应中断操作，如果在等待获取锁的过程中被打断则该方法会抛出InterruptedException异常。 lock.lockInterruptibly(); try { //如果count等数组长度则说明队列已满，当前线程将被挂起放到AQS队列中，等待队列非满时插入（非满条件）。 //在等待期间，锁会被释放，其他线程可以继续对队列进行操作。 while (count == items.length) notFull.await(); //如果队列可以存放元素，则调用enqueue将元素入队 enqueue(e); } finally { //释放锁 lock.unlock(); } }put方法内部调用了 enqueue 方法来实现元素入队，我们继续深入查看一下 enqueue 方法的实现细节：private void enqueue(E x) { //获取队列底层的数组 final Object[] items = this.items; //将putindex位置的值设置为我们传入的x items[putIndex] = x; //更新putindex，如果putindex等于数组长度，则更新为0 if (++putIndex == items.length) putIndex = 0; //队列长度+1 count++; //通知队列非空，那些因为获取元素而阻塞的线程可以继续工作了 notEmpty.signal(); }从源码中可以看到入队操作的逻辑就是在数组中追加一个新元素，整体执行步骤为:获取 ArrayBlockingQueue 底层的数组 items。将元素存到 putIndex 位置。更新 putIndex 到下一个位置，如果 putIndex 等于队列长度，则说明 putIndex 已经到达数组末尾了，下一次插入则需要 0 开始。(ArrayBlockingQueue 用到了循环队列的思想，即从头到尾循环复用一个数组)更新 count 的值，表示当前队列长度+1。调用 notEmpty.signal() 通知队列非空，消费者可以从队列中获取值了。自此我们了解了 put 方法的流程，为了更加完整的了解 ArrayBlockingQueue 关于生产者-消费者模型的设计，我们继续看看阻塞获取队列元素的 take 方法。public E take() throws InterruptedException { //获取锁 final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { //如果队列中元素个数为0，则将当前线程打断并存入AQS队列中，等待队列非空时获取并移除元素（非空条件） while (count == 0) notEmpty.await(); //如果队列不为空则调用dequeue获取元素 return dequeue(); } finally { //释放锁 lock.unlock(); } }理解了 put 方法再看take 方法就很简单了，其核心逻辑和put 方法正好是相反的，比如put 方法在队列满的时候等待队列非满时插入元素（非满条件），而take 方法等待队列非空时获取并移除元素（非空条件）。take方法内部调用了 dequeue 方法来实现元素出队，其核心逻辑和 enqueue 方法也是相反的。private E dequeue() { //获取阻塞队列底层的数组 final Object[] items = this.items; @SuppressWarnings("unchecked") //从队列中获取takeIndex位置的元素 E x = (E) items[takeIndex]; //将takeIndex置空 items[takeIndex] = null; //takeIndex向后挪动，如果等于数组长度则更新为0 if (++takeIndex == items.length) takeIndex = 0; //队列长度减1 count--; if (itrs != null) itrs.elementDequeued(); //通知那些被打断的线程当前队列状态非满，可以继续存放元素 notFull.signal(); return x; }由于dequeue 方法（出队）和上面介绍的 enqueue 方法（入队）的步骤大致类似，这里就不重复介绍了。为了帮助理解，我专门画了一张图来展示 notEmpty(非空) 和 notFull （非满）这两个条件对象是如何控制 ArrayBlockingQueue 的存和取的。消费者：当消费者从队列中 take 或者 poll 等操作取出一个元素之后，就会通知队列非满，此时那些等待非满的生产者就会被唤醒等待获取 CPU 时间片进行入队操作。生产者：当生产者将元素存到队列中后，就会触发通知队列非空，此时消费者就会被唤醒等待 CPU 时间片尝试获取元素。如此往复，两个条件对象就构成一个环路，控制着多线程之间的存和取。非阻塞式获取和新增元素ArrayBlockingQueue 非阻塞式获取和新增元素的方法为：offer(E e)：将元素插入队列尾部。如果队列已满，则该方法会直接返回 false，不会等待并阻塞线程。poll()：获取并移除队列头部的元素，如果队列为空，则该方法会直接返回 null，不会等待并阻塞线程。add(E e)：将元素插入队列尾部。如果队列已满则会抛出 IllegalStateException 异常，底层基于 offer(E e) 方法。remove()：移除队列头部的元素，如果队列为空则会抛出 NoSuchElementException 异常，底层基于 poll()。peek()：获取但不移除队列头部的元素，如果队列为空，则该方法会直接返回 null，不会等待并阻塞线程。先来看看 offer 方法，逻辑和 put 差不多，唯一的区别就是入队失败时不会阻塞当前线程，而是直接返回 false。public boolean offer(E e) { //确保插入的元素不为null checkNotNull(e); //获取锁 final ReentrantLock lock = this.lock; lock.lock(); try { //队列已满直接返回false if (count == items.length) return false; else { //反之将元素入队并直接返回true enqueue(e); return true; } } finally { //释放锁 lock.unlock(); } }poll 方法同理，获取元素失败也是直接返回空，并不会阻塞获取元素的线程。public E poll() { final ReentrantLock lock = this.lock; //上锁 lock.lock(); try { //如果队列为空直接返回null，反之出队返回元素值 return (count == 0) ? null : dequeue(); } finally { lock.unlock(); } }add 方法其实就是对于 offer 做了一层封装，如下代码所示，可以看到 add 会调用没有规定时间的 offer，如果入队失败则直接抛异常。public boolean add(E e) { return super.add(e); } public boolean add(E e) { //调用offer方法如果失败直接抛出异常 if (offer(e)) return true; else throw new IllegalStateException("Queue full"); }remove 方法同理，调用 poll，如果返回 null 则说明队列没有元素，直接抛出异常。public E remove() { E x = poll(); if (x != null) return x; else throw new NoSuchElementException(); }peek() 方法的逻辑也很简单，内部调用了 itemAt 方法。public E peek() { //加锁 final ReentrantLock lock = this.lock; lock.lock(); try { //当队列为空时返回 null return itemAt(takeIndex); } finally { //释放锁 lock.unlock(); } } //返回队列中指定位置的元素 @SuppressWarnings("unchecked") final E itemAt(int i) { return (E) items[i]; }指定超时时间内阻塞式获取和新增元素在 offer(E e) 和 poll() 非阻塞获取和新增元素的基础上，设计者提供了带有等待时间的 offer(E e, long timeout, TimeUnit unit) 和 poll(long timeout, TimeUnit unit) ，用于在指定的超时时间内阻塞式地添加和获取元素。 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) { //时间到了队列还是满的，则直接返回false if (nanos <= 0) return false; //阻塞nanos时间，等待非满 nanos = notFull.awaitNanos(nanos); } enqueue(e); return true; } finally { lock.unlock(); } }可以看到，带有超时时间的 offer 方法在队列已满的情况下，会等待用户所传的时间段，如果规定时间内还不能存放元素则直接返回 false。public E poll(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { //队列为空，循环等待，若时间到还是空的，则直接返回null while (count == 0) { if (nanos <= 0) return null; nanos = notEmpty.awaitNanos(nanos); } return dequeue(); } finally { lock.unlock(); } }同理，带有超时时间的 poll 也一样，队列为空则在规定时间内等待，若时间到了还是空的，则直接返回 null。判断元素是否存在ArrayBlockingQueue 提供了 contains(Object o) 来判断指定元素是否存在于队列中。public boolean contains(Object o) { //若目标元素为空，则直接返回 false if (o == null) return false; //获取当前队列的元素数组 final Object[] items = this.items; //加锁 final ReentrantLock lock = this.lock; lock.lock(); try { // 如果队列非空 if (count > 0) { final int putIndex = this.putIndex; //从队列头部开始遍历 int i = takeIndex; do { if (o.equals(items[i])) return true; if (++i == items.length) i = 0; } while (i != putIndex); } return false; } finally { //释放锁 lock.unlock(); } }ArrayBlockingQueue 获取和新增元素的方法对比为了帮助理解 ArrayBlockingQueue ，我们再来对比一下上面提到的这些获取和新增元素的方法。新增元素：方法队列满时处理方式方法返回值put(E e)线程阻塞，直到中断或被唤醒voidoffer(E e)直接返回 falsebooleanoffer(E e, long timeout, TimeUnit unit)指定超时时间内阻塞，超过规定时间还未添加成功则返回 falsebooleanadd(E e)直接抛出 IllegalStateException 异常boolean获取/移除元素：方法队列空时处理方式方法返回值take()线程阻塞，直到中断或被唤醒Epoll()返回 nullEpoll(long timeout, TimeUnit unit)指定超时时间内阻塞，超过规定时间还是空的则返回 nullEpeek()返回 nullEremove()直接抛出 NoSuchElementException 异常booleanArrayBlockingQueue 相关面试题ArrayBlockingQueue 是什么？它的特点是什么？ArrayBlockingQueue 是 BlockingQueue 接口的有界队列实现类，常用于多线程之间的数据共享，底层采用数组实现，从其名字就能看出来了。ArrayBlockingQueue 的容量有限，一旦创建，容量不能改变。为了保证线程安全，ArrayBlockingQueue 的并发控制采用可重入锁 ReentrantLock ，不管是插入操作还是读取操作，都需要获取到锁才能进行操作。并且，它还支持公平和非公平两种方式的锁访问机制，默认是非公平锁。ArrayBlockingQueue 虽名为阻塞队列，但也支持非阻塞获取和新增元素（例如 poll() 和 offer(E e) 方法），只是队列满时添加元素会抛出异常，队列为空时获取的元素为 null，一般不会使用。ArrayBlockingQueue 和 LinkedBlockingQueue 有什么区别？ArrayBlockingQueue 和 LinkedBlockingQueue 是 Java 并发包中常用的两种阻塞队列实现，它们都是线程安全的。不过，不过它们之间也存在下面这些区别：底层实现：ArrayBlockingQueue 基于数组实现，而 LinkedBlockingQueue 基于链表实现。是否有界：ArrayBlockingQueue 是有界队列，必须在创建时指定容量大小。LinkedBlockingQueue 创建时可以不指定容量大小，默认是Integer.MAX_VALUE，也就是无界的。但也可以指定队列大小，从而成为有界的。锁是否分离： ArrayBlockingQueue中的锁是没有分离的，即生产和消费用的是同一个锁；LinkedBlockingQueue中的锁是分离的，即生产用的是putLock，消费是takeLock，这样可以防止生产者和消费者线程之间的锁争夺。内存占用：ArrayBlockingQueue 需要提前分配数组内存，而 LinkedBlockingQueue 则是动态分配链表节点内存。这意味着，ArrayBlockingQueue 在创建时就会占用一定的内存空间，且往往申请的内存比实际所用的内存更大，而LinkedBlockingQueue 则是根据元素的增加而逐渐占用内存空间。ArrayBlockingQueue 和 ConcurrentLinkedQueue 有什么区别？ArrayBlockingQueue 和 ConcurrentLinkedQueue 是 Java 并发包中常用的两种队列实现，它们都是线程安全的。不过，不过它们之间也存在下面这些区别：底层实现：ArrayBlockingQueue 基于数组实现，而 ConcurrentLinkedQueue 基于链表实现。是否有界：ArrayBlockingQueue 是有界队列，必须在创建时指定容量大小，而 ConcurrentLinkedQueue 是无界队列，可以动态地增加容量。是否阻塞：ArrayBlockingQueue 支持阻塞和非阻塞两种获取和新增元素的方式（一般只会使用前者）， ConcurrentLinkedQueue 是无界的，仅支持非阻塞式获取和新增元素。ArrayBlockingQueue 的实现原理是什么？ArrayBlockingQueue 的实现原理主要分为以下几点（这里以阻塞式获取和新增元素为例介绍）：ArrayBlockingQueue 内部维护一个定长的数组用于存储元素。通过使用 ReentrantLock 锁对象对读写操作进行同步，即通过锁机制来实现线程安全。通过 Condition 实现线程间的等待和唤醒操作。这里再详细介绍一下线程间的等待和唤醒具体的实现（不需要记具体的方法，面试中回答要点即可）：当队列已满时，生产者线程会调用 notFull.await() 方法让生产者进行等待，等待队列非满时插入（非满条件）。当队列为空时，消费者线程会调用 notEmpty.await()方法让消费者进行等待，等待队列非空时消费（非空条件）。当有新的元素被添加时，生产者线程会调用 notEmpty.signal()方法唤醒正在等待消费的消费者线程。当队列中有元素被取出时，消费者线程会调用 notFull.signal()方法唤醒正在等待插入元素的生产者线程。关于 Condition接口的补充：Condition是 JDK1.5 之后才有的，它具有很好的灵活性，比如可以实现多路通知功能也就是在一个Lock对象中可以创建多个Condition实例（即对象监视器），线程对象可以注册在指定的Condition中，从而可以有选择性的进行线程通知，在调度线程上更加灵活。 在使用notify()/notifyAll()方法进行通知时，被通知的线程是由 JVM 选择的，用ReentrantLock类结合Condition实例可以实现“选择性通知” ，这个功能非常重要，而且是 Condition 接口默认提供的。而synchronized关键字就相当于整个 Lock 对象中只有一个Condition实例，所有的线程都注册在它一个身上。如果执行notifyAll()方法的话就会通知所有处于等待状态的线程，这样会造成很大的效率问题。而Condition实例的signalAll()方法，只会唤醒注册在该Condition实例中的所有等待线程。参考文献深入理解 Java 系列 | BlockingQueue 用法详解：https://juejin.cn/post/6999798721269465102深入浅出阻塞队列 BlockingQueue 及其典型实现 ArrayBlockingQueue：https://zhuanlan.zhihu.com/p/539619957并发编程大扫盲：ArrayBlockingQueue 底层原理和实战：https://zhuanlan.zhihu.com/p/339662987