一、CAS 无锁算法 public class ReentrantSpinLock { private AtomicReference<Thread> cas = new AtomicReference<Thread>(); private int count; public void lock() { Thread current = Thread.currentThread(); if (current == cas.get()) { // 如果当前线程已经获取到了锁，线程数增加一，然后返回 count++; return; } // 如果没获取到锁，则通过CAS自旋 while (!cas.compareAndSet(null, current)) { // DO nothing } } public void unlock() { Thread cur = Thread.currentThread(); if (cur == cas.get()) { if (count > 0) {// 如果大于0，表示当前线程多次获取了该锁，释放锁通过count减一来模拟 count--; } else {// 如果count==0，可以将锁释放，这样就能保证获取锁的次数与释放锁的次数是一致的了。 cas.compareAndSet(cur, null); } } } } TicketLock 解决公平性的问题。 public class TicketLockV2 { /** * 服务号 */ private AtomicInteger serviceNum = new AtomicInteger(); /** * 排队号 */ private AtomicInteger ticketNum = new AtomicInteger(); /** * 新增一个ThreadLocal，用于存储每个线程的排队号 */ private ThreadLocal<Integer> ticketNumHolder = new ThreadLocal<Integer>(); public void lock() { int currentTicketNum = ticketNum.getAndIncrement(); ticketNumHolder.set(currentTicketNum); System.out.println("LOCK " + Thread.currentThread().getName() + " currentTicketNum：" + currentTicketNum + " serviceNum：" + serviceNum.get()); while (currentTicketNum != serviceNum.get()) { Thread.onSpinWait(); // 添加自旋优化 } } public void unlock() { Integer currentTickNum = ticketNumHolder.get(); if (currentTickNum == null) { throw new IllegalMonitorStateException("应该先获取锁再释放锁"); } if (!serviceNum.compareAndSet(currentTickNum, currentTickNum + 1)) { throw new IllegalMonitorStateException("当前线程并未持有锁"); } System.out.println("UNLOCK " + Thread.currentThread().getName() + "- currentTicketNum：" + currentTickNum + "- serviceNum：" + serviceNum.get()); ticketNumHolder.remove(); // 清理ThreadLocal，防止内存泄漏 } } public class TicketLockV2Example { private static int counter = 0; private static final TicketLockV2 lock = new TicketLockV2(); public static void main(String[] args) throws InterruptedException { // 创建多个线程进行测试 int threadCount = 5; Thread[] threads = new Thread[threadCount]; for (int i = 0; i < threadCount; i++) { threads[i] = new Thread(() -> { lock.lock(); // 获取锁 counter++; // 临界区操作 lock.unlock(); // 确保在finally中释放锁 }, "Thread-" + i); } // 启动所有线程 for (Thread thread : threads) { thread.start(); } // 等待所有线程完成 for (Thread thread : threads) { thread.join(); } System.out.println("最终计数器的值: " + counter); } } 二、独占式锁 & 共享式锁 1、独占式锁（Exclusive Lock） public class ExclusiveLockExample { private final ReentrantLock lock = new ReentrantLock(); // 独占式锁 private int count = 0; public void write() { lock.lock(); // 只能被一个线程获取 try { count++; // 模拟写操作 Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + " 写入完成"); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } } } 2、共享式锁（Shared Lock） public class SharedLockExample { private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock(); private final Lock readLock = rwLock.readLock(); // 共享锁 private final Lock writeLock = rwLock.writeLock(); // 独占锁 private int data = 0; // 读操作：可以多个线程同时读 public void read() { readLock.lock(); // 获取共享锁 try { System.out.println(Thread.currentThread().getName() + " 读取数据: " + data); Thread.sleep(1000); // 模拟读取操作 } catch (InterruptedException e) { e.printStackTrace(); } finally { readLock.unlock(); } } // 写操作：只能一个线程写 public void write(int newData) { writeLock.lock(); // 获取独占锁 try { data = newData; System.out.println(Thread.currentThread().getName() + " 写入数据: " + newData); Thread.sleep(1000); // 模拟写入操作 } catch (InterruptedException e) { e.printStackTrace(); } finally { writeLock.unlock(); } } } 使用示例： public class LockDemo { public static void main(String[] args) { SharedLockExample sharedLock = new SharedLockExample(); // 创建多个读线程 for (int i = 0; i < 3; i++) { new Thread(() -> { while (true) { sharedLock.read(); } }, "Reader-" + i).start(); } // 创建写线程 new Thread(() -> { int value = 0; while (true) { sharedLock.write(value++); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } }, "Writer").start(); } } 独占式锁： 同一时刻只能有一个线程获取锁 适用于写操作频繁的场景 典型实现：ReentrantLock, synchronized 特点： 互斥访问 可重入 高一致性 共享式锁： 同一时刻可以有多个线程获取锁 适用于读多写少的场景 典型实现：ReentrantReadWriteLock 的读锁 特点： 并发读 读写互斥 写写互斥 三、AQS AQS (AbstractQueuedSynchronizer) 的核心概念。 3.1、AQS 的基本结构 public abstract class AbstractQueuedSynchronizer { // 同步状态 private volatile int state; // 等待队列的头节点 private transient volatile Node head; // 等待队列的尾节点 private transient volatile Node tail; // 内部类 Node，表示等待队列中的节点 static final class Node { volatile Node prev; // 前驱节点 volatile Node next; // 后继节点 volatile Thread thread; // 当前线程 volatile int waitStatus; // 等待状态 } } 3.2、基于 AQS 实现一个简单的互斥锁 public class SimpleLock extends AbstractQueuedSynchronizer { // 尝试获取锁 protected boolean tryAcquire(int arg) { // 期望状态为0，表示锁未被占用 if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } // 释放锁 protected boolean tryRelease(int arg) { if (getState() == 0) { throw new IllegalMonitorStateException(); } setExclusiveOwnerThread(null); setState(0); return true; } // 对外提供的加锁方法 public void lock() { acquire(1); } // 对外提供的解锁方法 public void unlock() { release(1); } } 3.3、AQS 的工作流程示例： public class AQSExample { private static final SimpleLock lock = new SimpleLock(); private static int count = 0; public static void main(String[] args) { // 创建多个线程竞争锁 for (int i = 0; i < 3; i++) { new Thread(() -> { lock.lock(); try { // 临界区操作 count++; System.out.println(Thread.currentThread().getName() + " 获得锁"); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } }, "Thread-" + i).start(); } } } 3.4、AQS 的核心机制（源码） // AQS 的主要操作流程 public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) { selfInterrupt(); } } // 入队操作 private Node addWaiter(Node mode) { // 创建新节点 Node node = new Node(Thread.currentThread(), mode); Node pred = tail; if (pred != null) { node.prev = pred; // CAS设置尾节点 if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 初始化队列 enq(node); return node; } 3.5、AQS 的主要特点 状态管理： // 获取当前同步状态 protected final int getState() { return state; } // 设置同步状态 protected final void setState(int newState) { state = newState; } // CAS 修改同步状态 protected final boolean compareAndSetState(int expect, int update) { return unsafe.compareAndSwapInt(this, stateOffset, expect, update); } 队列管理： // 等待队列示意 head -> Node1 -> Node2 -> Node3 -> tail // 每个节点包含： // - 线程引用 // - 等待状态 // - 前驱和后继指针 3.6、AQS 的实际应用 // ReentrantLock 的实现示例 public class ReentrantLock { private final Sync sync; // 公平锁 public static class FairSync extends Sync { protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { // 检查队列中是否有前驱节点 if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } // 处理重入 else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; setState(nextc); return true; } return false; } } } AQS 的核心概念： 同步状态（State）： 使用 volatile 修饰的整数 表示当前同步状态 通过 CAS 操作进行修改 等待队列： 双向链表结构 FIFO 顺序 支持独占和共享两种模式 实现方式： 独占模式：ReentrantLock 共享模式：Semaphore、CountDownLatch 组合模式：ReentrantReadWriteLock 主要优点： 提供了标准化的同步器实现框架 简化了锁的实现 支持公平和非公平两种方式 使用 AQS 的最佳实践： 继承 AQS 实现自定义同步器时： 重写 tryAcquire/tryRelease（独占模式） 重写 tryAcquireShared/tryReleaseShared（共享模式） 根据需要实现 isHeldExclusively 注意事项： 正确维护同步状态 考虑公平性需求 处理中断和超时 四、 ConcurrentHashMap & HashMap 4.1、基本结构 // HashMap public class HashMap<K,V> { // 数组 + 链表/红黑树 Node<K,V>[] table; static class Node<K,V> { final int hash; final K key; V value; Node<K,V> next; } } // ConcurrentHashMap public class ConcurrentHashMap<K,V> { // 数组 + 链表/红黑树 + 分段锁 private transient volatile Node<K,V>[] table; static class Node<K,V> { final int hash; final K key; volatile V value; volatile Node<K,V> next; } } 4.2、并发操作对比 // HashMap 线程不安全的示例 public class HashMapThreadUnsafe { public static void main(String[] args) throws InterruptedException { Map<Integer, Integer> map = new HashMap<>(); // 多线程并发写入 Thread t1 = new Thread(() -> { for (int i = 0; i < 1000; i++) { map.put(i, i); } }); Thread t2 = new Thread(() -> { for (int i = 1000; i < 2000; i++) { map.put(i, i); } }); t1.start(); t2.start(); t1.join(); t2.join(); // 结果可能小于2000 System.out.println("Size: " + map.size()); } } // ConcurrentHashMap 线程安全的示例 public class ConcurrentHashMapThreadSafe { public static void main(String[] args) throws InterruptedException { ConcurrentHashMap<Integer, Integer> map = new ConcurrentHashMap<>(); // 多线程并发写入 Thread t1 = new Thread(() -> { for (int i = 0; i < 1000; i++) { map.put(i, i); } }); Thread t2 = new Thread(() -> { for (int i = 1000; i < 2000; i++) { map.put(i, i); } }); t1.start(); t2.start(); t1.join(); t2.join(); // 结果一定是2000 System.out.println("Size: " + map.size()); } } 4.3、常见操作的实现差异 // 1. 插入操作 public class MapOperations { private HashMap<String, String> hashMap = new HashMap<>(); private ConcurrentHashMap<String, String> concurrentHashMap = new ConcurrentHashMap<>(); // HashMap 插入 public void putInHashMap(String key, String value) { hashMap.put(key, value); // 非线程安全 } // ConcurrentHashMap 插入 public void putInConcurrentHashMap(String key, String value) { concurrentHashMap.put(key, value); // 线程安全 } // ConcurrentHashMap 原子操作 public void atomicOperation() { // 原子性的 put-if-absent concurrentHashMap.putIfAbsent("key", "value"); // 原子性的 replace concurrentHashMap.replace("key", "oldValue", "newValue"); // 原子性的计算 concurrentHashMap.compute("key", (k, v) -> v == null ? "default" : v + "updated"); } } 4.4、实际应用场景 // 场景1：缓存实现 public class CacheImplementation { private ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>(); public Object getFromCache(String key, Supplier<Object> valueLoader) { return cache.computeIfAbsent(key, k -> valueLoader.get()); } } // 场景2：计数器 public class ConcurrentCounter { private ConcurrentHashMap<String, AtomicInteger> counters = new ConcurrentHashMap<>(); public void increment(String key) { counters.computeIfAbsent(key, k -> new AtomicInteger(0)).incrementAndGet(); } } // 场景3：会话管理 public class SessionManager { private ConcurrentHashMap<String, UserSession> sessions = new ConcurrentHashMap<>(); public void addSession(String sessionId, UserSession session) { sessions.put(sessionId, session); } public UserSession getSession(String sessionId) { return sessions.get(sessionId); } } 4.5、性能优化场景 public class PerformanceOptimization { private ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>(); // 批量操作 public void batchOperation() { // 使用 forEach 进行并行处理 map.forEach((k, v) -> { // 处理每个元素 processEntry(k, v); }); // 使用 search 查找元素 String result = map.search(1, (k, v) -> { if (someCondition(k, v)) { return v; } return null; }); } } 4.6、总结 主要区别： 线程安全性： HashMap：非线程安全 ConcurrentHashMap：线程安全 实现机制： HashMap：数组 + 链表/红黑树 ConcurrentHashMap：数组 + 链表/红黑树 + CAS + synchronized 性能特点： HashMap：单线程性能好 ConcurrentHashMap：多线程性能好 使用场景： HashMap 适用于： 单线程环境 读多写少 对性能要求高 数据量较小 ConcurrentHashMap 适用于： 多线程环境 并发读写 需要线程安全 缓存实现 计数器 会话管理 注意事项： 容量规划： 初始容量设置 负载因子选择 并发级别考虑 性能优化： 避免过度同步 合理使用批量操作 注意内存使用 功能选择： 根据实际需求选择 考虑是否需要线程安全 评估性能要求
一、简介 CountDownLatch 是一个同步工具类，它通过一个计数器来实现的，初始值为线程的数量。每当一个线程完成了自己的任务，计数器的值就相应得减 1。当计数器到达 0 时，表示所有的线程都已执行完毕，然后在等待的线程就可以恢复执行任务。 二、方法详解 1、CountDownLatch (int count)：count 为计数器的初始值（一般需要多少个线程执行，count 就设为几）。 2、countDown (): 每调用一次计数器值 - 1，直到 count 被减为 0，代表所有线程全部执行完毕。 3、getCount ()：获取当前计数器的值。 4、await (): 等待计数器变为 0，即等待所有异步线程执行完毕。 5、boolean await (long timeout, TimeUnit unit)： 此方法与 await () 区别： 此方法至多会等待指定的时间，超时后会自动唤醒，若 timeout 小于等于零，则不会等待 boolean 类型返回值：若计数器变为零了，则返回 true；若指定的等待时间过去了，则返回 false 三、CountDownLatch 应用场景 1、某个线程需要在其他 n 个线程执行完毕后再向下执行。 2、多个线程并行执行同一个任务，提高响应速度。 四、CountDownLatch 代码示例 public class CountDownLaunchTest { public static void main(String[] args) throws InterruptedException { List list = new ArrayList(); for(int i = 1; i<=100; i++){ list.add(i); } Long start = System.currentTimeMillis(); for(int i = 0; i<list.size(); i++){ Thread.sleep(100); } System.out.println("=====同步执行：耗时"+(System.currentTimeMillis()-start)+"毫秒======"); Long start1 = System.currentTimeMillis(); CountDownLatch latch = new CountDownLatch(10); for(int i = 0; i<latch.getCount(); i++){ new Thread(new Test(latch, i, list)).start(); } latch.await(); System.out.println("=====异步执行：耗时"+(System.currentTimeMillis()-start1)+"毫秒======"); } static class Test implements Runnable{ private CountDownLatch latch; private int i; private List list; Test(CountDownLatch latch, int i, List list){ this.latch = latch; this.i = i; this.list = list; } @SneakyThrows @Override public void run() { for(int a = i*10; a<(i+1)*10; a++){ // 执行任务逻辑 Thread.sleep(100); } latch.countDown(); } } }
一、简介 在实际编程中，会经常使用到 JDK 中 Collection 集合框架中的各种容器类如实现 List,Map,Queue 接口的容器类，但是这些容器类基本上不是线程安全的，除了使用 Collections 可以将其转换为线程安全的容器，Doug Lea 大师为我们都准备了对应的线程安全的容器，如实现 List 接口的 CopyOnWriteArrayList，实现 Map 接口的 ConcurrentHashMap， 实现 Queue 接口的 ConcurrentLinkedQueue。 最常用的 生产者 - 消费者 问题中，队列通常被视作线程间操作的数据容器，这样，可以对各个模块的业务功能进行解耦，生产者将 “生产” 出来的数据放置在数据容器中，而消费者仅仅只需要在 “数据容器” 中进行获取数据即可，这样生产者线程和消费者线程就能够进行解耦，只专注于自己的业务功能即可。阻塞队列（BlockingQueue）被广泛使用在 “生产者 - 消费者” 问题中，其原因是 BlockingQueue 提供了可阻塞的插入和移除的方法。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。 二、基本操作 BlockingQueue 基本操作总结如下： BlockingQueue 继承于 Queue 接口，因此，对数据元素的基本操作有： 插入元素 add (E e) ：往队列插入数据，当队列满时，插入元素时会抛出 IllegalStateException 异常； offer (E e)：当往队列插入数据时，插入成功返回 `true`，否则则返回 `false`。当队列满时不会抛出异常； 删除元素 take ()：当阻塞队列为空时，获取队头数据的线程会被阻塞； poll (long timeout, TimeUnit unit)：当阻塞队列为空时，获取数据的线程会被阻塞，另外，如果被阻塞的线程超过了给定的时长，该线程会退出 三、常用的 BlockingQueue 实现 BlockingQueue 接口的有 ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, LinkedTransferQueue, PriorityBlockingQueue, SynchronousQueue， 而这几种常见的阻塞队列也是在实际编程中会常用的，下面对这几种常见的阻塞队列进行说明。 3.1、ArrayBlockingQueue ArrayBlockingQueue 是由数组实现的有界阻塞队列。该队列命令元素 FIFO（先进先出）。因此，对头元素时队列中存在时间最长的数据元素，而对尾数据则是当前队列最新的数据元素。ArrayBlockingQueue 可作为 “有界数据缓冲区”，生产者插入数据到队列容器中，并由消费者提取。ArrayBlockingQueue 一旦创建，容量不能改变。 当队列容量满时，尝试将元素放入队列将导致操作阻塞；尝试从一个空队列中取一个元素也会同样阻塞。 ArrayBlockingQueue 默认情况下不能保证线程访问队列的公平性，所谓公平性是指严格按照线程等待的绝对时间顺序，即最先等待的线程能够最先访问到 ArrayBlockingQueue。而非公平性则是指访问 ArrayBlockingQueue 的顺序不是遵守严格的时间顺序，有可能存在，一旦 ArrayBlockingQueue 可以被访问时，长时间阻塞的线程依然无法访问到 ArrayBlockingQueue。如果保证公平性，通常会降低吞吐量。如果需要获得公平性的 ArrayBlockingQueue，可采用如下代码： private static ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(10,true); 3.2、LinkedBlockingQueue LinkedBlockingQueue 是用链表实现的有界阻塞队列，同样满足 FIFO 的特性，与 ArrayBlockingQueue 相比起来具有更高的吞吐量，为了防止 LinkedBlockingQueue 容量迅速增，损耗大量内存。通常在创建 LinkedBlockingQueue 对象时，会指定其大小，如果未指定，容量等于 Integer.MAX_VALUE。 3.3、PriorityBlockingQueue PriorityBlockingQueue 是一个支持优先级的无界阻塞队列。默认情况下元素采用自然顺序进行排序，也可以通过自定义类实现 compareTo () 方法来指定元素排序规则，或者初始化时通过构造器参数 Comparator 来指定排序规则。 3.4、SynchronousQueue SynchronousQueue 每个插入操作必须等待另一个线程进行相应的删除操作，因此，SynchronousQueue 实际上没有存储任何数据元素，因为只有线程在删除数据时，其他线程才能插入数据，同样的，如果当前有线程在插入数据时，线程才能删除数据。SynchronousQueue 也可以通过构造器参数来为其指定公平性。 3.5、LinkedTransferQueue LinkedTransferQueue 是一个由链表数据结构构成的无界阻塞队列，由于该队列实现了 TransferQueue 接口，与其他阻塞队列相比主要有以下不同的方法： transfer(E e) 如果当前有线程（消费者）正在调用 take () 方法或者可延时的 poll () 方法进行消费数据时，生产者线程可以调用 transfer 方法将数据传递给消费者线程。如果当前没有消费者线程的话，生产者线程就会将数据插入到队尾，直到有消费者能够进行消费才能退出； tryTransfer(E e) tryTransfer 方法如果当前有消费者线程（调用 take 方法或者具有超时特性的 poll 方法）正在消费数据的话，该方法可以将数据立即传送给消费者线程，如果当前没有消费者线程消费数据的话，就立即返回 false。因此，与 transfer 方法相比，transfer 方法是必须等到有消费者线程消费数据时，生产者线程才能够返回。而 tryTransfer 方法能够立即返回结果退出。 tryTransfer(E e,long timeout,imeUnit unit) 与 transfer 基本功能一样，只是增加了超时特性，如果数据才规定的超时时间内没有消费者进行消费的话，就返回 false。 3.6、LinkedBlockingDeque LinkedBlockingDeque 是基于链表数据结构的有界阻塞双端队列，如果在创建对象时为指定大小时，其默认大小为 Integer.MAX_VALUE。与 LinkedBlockingQueue 相比，主要的不同点在于，LinkedBlockingDeque 具有双端队列的特性。LinkedBlockingDeque 基本操作如下图所示： 如上图所示，LinkedBlockingDeque 的基本操作可以分为四种类型：1. 特殊情况，抛出异常；2. 特殊情况，返回特殊值如 null 或者 false；3. 当线程不满足操作条件时，线程会被阻塞直至条件满足；4. 操作具有超时特性。 另外，LinkedBlockingDeque 实现了 BlockingDueue 接口而 LinkedBlockingQueue 实现的是 BlockingQueue，这两个接口的主要区别如下图所示： 从上图可以看出，两个接口的功能是可以等价使用的，比如 BlockingQueue 的 add 方法和 BlockingDeque 的 addLast 方法的功能是一样的。 3.7、DelayQueue DelayQueue 是一个存放实现 Delayed 接口的数据的无界阻塞队列，只有当数据对象的延时时间达到时才能插入到队列进行存储。如果当前所有的数据都还没有达到创建时所指定的延时期，则队列没有队头，并且线程通过 poll 等方法获取数据元素则返回 null。所谓数据延时期满时，则是通过 Delayed 接口的 getDelay (TimeUnit.NANOSECONDS) 来进行判定，如果该方法返回的是小于等于 0 则说明该数据元素的延时期已满。 3.7.1、阻塞队列 借助同步阻塞队列，也可以实现并发控制的效果，比如队列中初始化 n 个元素，每次消费从队列中获取一个元素，如果拿不到则阻塞；执行完毕之后，重新塞入一个元素，这样就可以实现一个简单版的并发控制。 public class BlockQueueTest { AtomicInteger cnt = new AtomicInteger(); private void consumer(LinkedBlockingQueue<Integer> queue) { try { // 同步阻塞拿去数据 int val = queue.take(); Thread.sleep(2000); System.out.println("成功拿到: " + val + " Thread: " + Thread.currentThread()); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 添加数据 System.out.println("结束 " + Thread.currentThread()); queue.offer(cnt.getAndAdd(1)); } } @Test public void blockQueue() throws InterruptedException { LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(2); queue.add(cnt.getAndAdd(1)); queue.add(cnt.getAndAdd(1)); new Thread(() -> consumer(queue)).start(); new Thread(() -> consumer(queue)).start(); new Thread(() -> consumer(queue)).start(); new Thread(() -> consumer(queue)).start(); Thread.sleep(10000); } } 3.7.2、信号量 Semaphore 队列的实现方式，可以使用信号量 Semaphore 来完成，通过设置信号量，来控制并发数。 public class SemaphoreTest { private void semConsumer(Semaphore semaphore) { try { // 同步阻塞，尝试获取信号 semaphore.acquire(1); System.out.println("成功拿到信号，执行: " + Thread.currentThread().getName()); Thread.sleep(2000); System.out.println("执行完毕，释放信号: " + Thread.currentThread().getName()); semaphore.release(1); } catch (Exception e) { e.printStackTrace(); } } @Test public void semaphore() throws InterruptedException { Semaphore semaphore = new Semaphore(2); new Thread(() -> semConsumer(semaphore)).start(); new Thread(() -> semConsumer(semaphore)).start(); new Thread(() -> semConsumer(semaphore)).start(); new Thread(() -> semConsumer(semaphore)).start(); new Thread(() -> semConsumer(semaphore)).start(); Thread.sleep(20_000); } } 3.7.3、计数器 CountDownLatch 计数，应用场景更偏向于多线程的协同，比如多个线程执行完毕之后，再处理某些事情；不同于上面的并发数的控制，它和栅栏一样，更多的是行为结果的统一。 public class CountDownLatchTest { @Test public void countDown() throws InterruptedException { CountDownLatch countDownLatch = new CountDownLatch(2); new Thread(() -> { try { System.out.println("do something in " + Thread.currentThread()); Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } finally { countDownLatch.countDown(); } }).start(); new Thread(() -> { try { System.out.println("do something in t2: " + Thread.currentThread()); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { countDownLatch.countDown(); } }).start(); countDownLatch.await(); System.out.printf("结束"); } } 3.7.4、栅栏 CyclicBarrier CyclicBarrier 的作用与上面的 CountDownLatch 相似，区别在于正向计数 + 1, 只有达到条件才放行；且支持通过调用 reset () 重置计数，而 CountDownLatch 则不行。 public class CyclicBarrierTest { private void cyclicBarrierLogic(CyclicBarrier barrier, long sleep) { // 等待达到条件才放行 try { System.out.println("准备执行: " + Thread.currentThread().getName() + " at: " + LocalDateTime.now()); Thread.sleep(sleep); int index = barrier.await(); System.out.println("开始执行: " + index + " thread: " + Thread.currentThread() + " at: " + LocalDateTime.now()); } catch (Exception e) { e.printStackTrace(); } } @Test public void testCyclicBarrier() throws InterruptedException { // 到达两个工作线程才能继续往后面执行 CyclicBarrier barrier = new CyclicBarrier(2); // 三秒之后，下面两个线程的才会输出 开始执行 new Thread(() -> cyclicBarrierLogic(barrier, 1000)).start(); new Thread(() -> cyclicBarrierLogic(barrier, 3000)).start(); Thread.sleep(4000); // 重置，可以再次使用 barrier.reset(); new Thread(() -> cyclicBarrierLogic(barrier, 1)).start(); new Thread(() -> cyclicBarrierLogic(barrier, 1)).start(); Thread.sleep(10000); } } 3.7.5、guava令牌桶 guava 封装了非常简单的并发控制工具类 RateLimiter，作为单机的并发控制首选。 public class GuavaTest { private void guavaProcess(RateLimiter rateLimiter) { try { // 同步阻塞方式获取 System.out.println("准备执行: " + Thread.currentThread().getName() + " > " + LocalDateTime.now()); rateLimiter.acquire(); System.out.println("执行中: " + Thread.currentThread().getName() + " > " + LocalDateTime.now()); } catch (Exception e) { e.printStackTrace(); } } @Test public void testGuavaRate() throws InterruptedException { // 1s 中放行两个请求 RateLimiter rateLimiter = RateLimiter.create(2.0d); new Thread(() -> guavaProcess(rateLimiter)).start(); new Thread(() -> guavaProcess(rateLimiter)).start(); new Thread(() -> guavaProcess(rateLimiter)).start(); new Thread(() -> guavaProcess(rateLimiter)).start(); new Thread(() -> guavaProcess(rateLimiter)).start(); new Thread(() -> guavaProcess(rateLimiter)).start(); new Thread(() -> guavaProcess(rateLimiter)).start(); Thread.sleep(20_000); } }
一、原因 threadLocal 是为了解决对象不能被多线程共享访问的问题，通过 threadLocal.set 方法将对象实例保存在每个线程自己所拥有的 threadLocalMap 中，这样每个线程使用自己的对象实例，彼此不会影响达到隔离的作用，从而就解决了对象在被共享访问带来线程安全问题。如果将同步机制和 threadLocal 做一个横向比较的话，同步机制就是通过控制线程访问共享对象的顺序，而 threadLocal 就是为每一个线程分配一个该对象，各用各的互不影响。打个比方说，现在有 100 个同学需要填写一张表格但是只有一支笔，同步就相当于 A 使用完这支笔后给 B，B 使用后给 C 用...... 老师就控制着这支笔的使用顺序，使得同学之间不会产生冲突。而 threadLocal 就相当于，老师直接准备了 100 支笔，这样每个同学都使用自己的，同学之间就不会产生冲突。很显然这就是两种不同的思路，同步机制以 “时间换空间”，由于每个线程在同一时刻共享对象只能被一个线程访问造成整体上响应时间增加，但是对象只占有一份内存，牺牲了时间效率换来了空间效率即 时间换空间。而 threadLocal，为每个线程都分配了一份对象，自然而然内存使用率增加，每个线程各用各的，整体上时间效率要增加很多，牺牲了空间效率换来时间效率即 空间换时间。 threadLocal threadLocalMap entry 之间的关系如下图所示： 上图中，实线代表强引用，虚线代表的是弱引用，如果 threadLocal 外部强引用被置为 null (threadLocalInstance=null) 的话，threadLocal 实例就没有一条引用链路可达，很显然在 gc (垃圾回收) 的时候势必会被回收，因此 entry 就存在 key 为 null 的情况，无法通过一个 Key 为 null 去访问到该 entry 的 value。同时，就存在了这样一条引用链：threadRef->currentThread->threadLocalMap->entry->valueRef->valueMemory, 导致在垃圾回收的时候进行可达性分析的时候，value 可达从而不会被回收掉，但是该 value 永远不能被访问到，这样就存在了内存泄漏。当然，如果线程执行结束后，threadLocal，threadRef 会断掉，因此 threadLocal, threadLocalMap, entry 都会被回收掉。可是，在实际使用中我们都是会用线程池去维护我们的线程，比如在 Executors.newFixedThreadPool () 时创建线程的时候，为了复用线程是不会结束的，所以 threadLocal 内存泄漏就值得关注。 二、已经存在的改进 实际上，为了解决 threadLocal 潜在的内存泄漏的问题，Josh Bloch and Doug Lea 大师已经做了一些改进。在 threadLocal 的 set 和 get 方法中都有相应的处理。下文为了叙述，针对 key 为 null 的 entry，源码注释为 stale entry，直译为不新鲜的 entry，这里就称之为 “脏 entry”。比如在 ThreadLocalMap 的 set 方法中： private void set(ThreadLocal<?> key, Object value) { // We don't use a fast path as with get() because it is at // least as common to use set() to create new entries as // it is to replace existing ones, in which case, a fast // path would fail more often than not. 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(); } 在该方法中针对脏 entry 做了这样的处理： 1、如果当前 table[i]！=null 的话说明 hash 冲突就需要向后环形查找，若在查找过程中遇到脏 entry 就通过 replaceStaleEntry 进行处理。 2、如果当前 table[i]==null 的话说明新的 entry 可以直接插入，但是插入后会调用 cleanSomeSlots 方法检测并清除脏 entry。 2.1、cleanSomeSlots 该方法的源码为： /* @param i a position known NOT to hold a stale entry. The * scan starts at the element after i. * * @param n scan control: {@code log2(n)} cells are scanned, * unless a stale entry is found, in which case * {@code log2(table.length)-1} additional cells are scanned. * When called from insertions, this parameter is the number * of elements, but when from replaceStaleEntry, it is the * table length. (Note: all this could be changed to be either * more or less aggressive by weighting n instead of just * using straight log n. But this version is simple, fast, and * seems to work well.) * * @return true if any stale entries have been removed. */ 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; } 1、i 表示：插入 entry 的位置 i，很显然在上述情况 2（table[i]==null）中，entry 刚插入后该位置 i 很显然不是脏 entry; 2、参数 n 主要用于扫描控制（scan control），从 while 中是通过 n 来进行条件判断的说明 n 就是用来控制扫描趟数（循环次数）的。在扫描过程中，如果没有遇到脏 entry 就整个扫描过程持续 log2 (n) 次，log2 (n) 的得来是因为 n >>>= 1，每次 n 右移一位相当于 n 除以 2。如果在扫描过程中遇到脏 entry 的话就会令 n 为当前 hash 表的长度（n=len），再扫描 log2 (n) 趟，注意此时 n 增加无非就是多增加了循环次数从而通过 nextIndex 往后搜索的范围扩大，示意图如下 按照 n 的初始值，搜索范围为黑线，当遇到了脏 entry，此时 n 变成了哈希数组的长度（n 取值增大），搜索范围 log2 (n) 增大，红线表示。如果在整个搜索过程没遇到脏 entry 的话，搜索结束，采用这种方式的主要是用于时间效率上的平衡。 n 的取值 如果是在 set 方法插入新的 entry 后调用（上述情况 2），n 位当前已经插入的 entry 个数 size；如果是在 replaceSateleEntry 方法中调用 n 为哈希表的长度 len。 2.2、expungeStaleEntry 如果对输入参数能够理解的话，那么 cleanSomeSlots 方法搜索基本上清除了，但是全部搞定还需要掌握 expungeStaleEntry 方法，当在搜索过程中遇到了脏 entry 的话就会调用该方法去清理掉脏 entry。源码为： /** * Expunge a stale entry by rehashing any possibly colliding entries * lying between staleSlot and the next null slot. This also expunges * any other stale entries encountered before the trailing null. See * Knuth, Section 6.4 * * @param staleSlot index of slot known to have null key * @return the index of the next null slot after staleSlot * (all between staleSlot and this slot will have been checked * for expunging). */ private int expungeStaleEntry(int staleSlot) { Entry[] tab = table; int len = tab.length; //清除当前脏entry // expunge entry at staleSlot tab[staleSlot].value = null; tab[staleSlot] = null; size--; // Rehash until we encounter null Entry e; int i; //2.往后环形继续查找,直到遇到table[i]==null时结束 for (i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); //3. 如果在向后搜索过程中再次遇到脏entry，同样将其清理掉 if (k == null) { e.value = null; tab[i] = null; size--; } else { //处理rehash的情况 int h = k.threadLocalHashCode & (len - 1); if (h != i) { tab[i] = null; // Unlike Knuth 6.4 Algorithm R, we must scan until // null because multiple entries could have been stale. while (tab[h] != null) h = nextIndex(h, len); tab[h] = e; } } } return i; } 该方法逻辑请看注释（第 1,2,3 步），主要做了这么几件事情： 1、清理当前脏 entry，即将其 value 引用置为 null，并且将 table [staleSlot] 也置为 null。value 置为 null 后该 value 域变为不可达，在下一次 gc 的时候就会被回收掉，同时 table [staleSlot] 为 null 后以便于存放新的 entry; 2、从当前 staleSlot 位置向后环形（nextIndex）继续搜索，直到遇到哈希桶（tab [i]）为 null 的时候退出； 3、若在搜索过程再次遇到脏 entry，继续将其清除。 也就是说该方法，清理掉当前脏 entry 后，并没有闲下来继续向后搜索，若再次遇到脏 entry 继续将其清理，直到哈希桶（table [i]）为 null 时退出。因此方法执行完的结果为 从当前脏 entry（staleSlot）位到返回的 i 位，这中间所有的 entry 不是脏 entry。为什么是遇到 null 退出呢？原因是存在脏 entry 的前提条件是 当前哈希桶（table [i]）不为 null, 只是该 entry 的 key 域为 null。如果遇到哈希桶为 null, 很显然它连成为脏 entry 的前提条件都不具备。 现在对 cleanSomeSlot 方法做一下总结，其方法执行示意图如下： 如图所示，cleanSomeSlot 方法主要有这样几点： 1、从当前位置 i 处（位于 i 处的 entry 一定不是脏 entry）为起点在初始小范围（log2 (n)，n 为哈希表已插入 entry 的个数 size）开始向后搜索脏 entry，若在整个搜索过程没有脏 entry，方法结束退出 2、如果在搜索过程中遇到脏 entryt 通过 expungeStaleEntry 方法清理掉当前脏 entry，并且该方法会返回下一个哈希桶 (table [i]) 为 null 的索引位置为 i。这时重新令搜索起点为索引位置 i，n 为哈希表的长度 len，再次扩大搜索范围为 log2 (n') 继续搜索。 下面，以一个例子更清晰的来说一下，假设当前 table 数组的情况如下图。 1、如图当前 n 等于 hash 表的 size 即 n=10，i=1, 在第一趟搜索过程中通过 nextIndex,i 指向了索引为 2 的位置，此时 table 2 为 null，说明第一趟未发现脏 entry, 则第一趟结束进行第二趟的搜索。 2、第二趟所搜先通过 nextIndex 方法，索引由 2 的位置变成了 i=3, 当前 table 3!=null 但是该 entry 的 key 为 null，说明找到了一个脏 entry，先将 n 置为哈希表的长度 len, 然后继续调用 expungeStaleEntry 方法，该方法会将当前索引为 3 的脏 entry 给清除掉（令 value 为 null，并且 table 3 也为 null）, 但是该方法可不想偷懒，它会继续往后环形搜索，往后会发现索引为 4,5 的位置的 entry 同样为脏 entry，索引为 6 的位置的 entry 不是脏 entry 保持不变，直至 i=7 的时候此处 table 7 位 null，该方法就以 i=7 返回。至此，第二趟搜索结束； 3、由于在第二趟搜索中发现脏 entry，n 增大为数组的长度 len，因此扩大搜索范围（增大循环次数）继续向后环形搜索； 4、直到在整个搜索范围里都未发现脏 entry，cleanSomeSlot 方法执行结束退出。 2.3、replaceStaleEntry 先来看 replaceStaleEntry 方法，该方法源码为： /* * @param key the key * @param value the value to be associated with key * @param staleSlot index of the first stale entry encountered while * searching for key. */ private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) { Entry[] tab = table; int len = tab.length; Entry e; // Back up to check for prior stale entry in current run. // We clean out whole runs at a time to avoid continual // incremental rehashing due to garbage collector freeing // up refs in bunches (i.e., whenever the collector runs). //向前找到第一个脏entry int slotToExpunge = staleSlot; for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)) if (e.get() == null) slotToExpunge = i; // Find either the key or trailing null slot of run, whichever // occurs first for (int i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) { ThreadLocal<?> k = e.get(); // If we find key, then we need to swap it // with the stale entry to maintain hash table order. // The newly stale slot, or any other stale slot // encountered above it, can then be sent to expungeStaleEntry // to remove or rehash all of the other entries in run. if (k == key) { //如果在向后环形查找过程中发现key相同的entry就覆盖并且和脏entry进行交换 e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e; // Start expunge at preceding stale entry if it exists //如果在查找过程中还未发现脏entry，那么就以当前位置作为cleanSomeSlots //的起点 if (slotToExpunge == staleSlot) slotToExpunge = i; //搜索脏entry并进行清理 cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; } // If we didn't find stale entry on backward scan, the // first stale entry seen while scanning for key is the // first still present in the run. //如果向前未搜索到脏entry，则在查找过程遇到脏entry的话，后面就以此时这个位置 //作为起点执行cleanSomeSlots if (k == null && slotToExpunge == staleSlot) slotToExpunge = i; } // If key not found, put new entry in stale slot //如果在查找过程中没有找到可以覆盖的entry，则将新的entry插入在脏entry tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value); // If there are any other stale entries in run, expunge them if (slotToExpunge != staleSlot) //执行cleanSomeSlots cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); } 该方法的逻辑请看注释，下面我结合各种情况详细说一下该方法的执行过程。首先先看这一部分的代码： int slotToExpunge = staleSlot; for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len)) if (e.get() == null) slotToExpunge = i; 这部分代码通过 PreIndex 方法实现往前环形搜索脏 entry 的功能，初始时 slotToExpunge 和 staleSlot 相同，若在搜索过程中发现了脏 entry，则更新 slotToExpunge 为当前索引 i。另外，说明 replaceStaleEntry 并不仅仅局限于处理当前已知的脏 entry，它认为在出现脏 entry 的相邻位置也有很大概率出现脏 entry，所以为了一次处理到位，就需要向前环形搜索，找到前面的脏 entry。那么根据在向前搜索中是否还有脏 entry 以及在 for 循环后向环形查找中是否找到可覆盖的 entry，我们分这四种情况来充分理解这个方法： 1、前向有脏 entry 1.1、后向环形查找找到可覆盖的 entry 该情形如下图所示。 如图，slotToExpunge 初始状态和 staleSlot 相同，当前向环形搜索遇到脏 entry 时，在第 1 行代码中 slotToExpunge 会更新为当前脏 entry 的索引 i，直到遇到哈希桶（table [i]）为 null 的时候，前向搜索过程结束。在接下来的 for 循环中进行后向环形查找，若查找到了可覆盖的 entry，第 2,3,4 行代码先覆盖当前位置的 entry，然后再与 staleSlot 位置上的脏 entry 进行交换。交换之后脏 entry 就更换到了 i 处，最后使用 cleanSomeSlots 方法从 slotToExpunge 为起点开始进行清理脏 entry 的过程。 1.2、后向环形查找未找到可覆盖的 entry 该情形如下图所示。 2、前向没有脏 entry 2.1、后向环形查找找到可覆盖的 entry 该情形如下图所示。 如图，slotToExpunge 初始状态和 staleSlot 相同，当前向环形搜索直到遇到哈希桶（table [i]）为 null 的时候，前向搜索过程结束，若在整个过程未遇到脏 entry，slotToExpunge 初始状态依旧和 staleSlot 相同。在接下来的 for 循环中进行后向环形查找，若遇到了脏 entry，在第 7 行代码中更新 slotToExpunge 为位置 i。若查找到了可覆盖的 entry，第 2,3,4 行代码先覆盖当前位置的 entry，然后再与 staleSlot 位置上的脏 entry 进行交换，交换之后脏 entry 就更换到了 i 处。如果在整个查找过程中都还没有遇到脏 entry 的话，会通过第 5 行代码，将 slotToExpunge 更新当前 i 处，最后使用 cleanSomeSlots 方法从 slotToExpunge 为起点开始进行清理脏 entry 的过程。 2.2、后向环形查找未找到可覆盖的 entry 该情形如下图所示。 如图，slotToExpunge 初始状态和 staleSlot 相同，当前向环形搜索直到遇到哈希桶（table [i]）为 null 的时候，前向搜索过程结束，若在整个过程未遇到脏 entry，slotToExpunge 初始状态依旧和 staleSlot 相同。在接下来的 for 循环中进行后向环形查找，若遇到了脏 entry，在第 7 行代码中更新 slotToExpunge 为位置 i。若没有查找到了可覆盖的 entry，哈希桶（table [i]）为 null 的时候，后向环形查找过程结束。那么接下来在 8,9 行代码中，将插入的新 entry 直接放在 staleSlot 处即可。另外，如果发现 slotToExpunge 被重置，则第 10 行代码 if 判断为 true, 就使用 cleanSomeSlots 方法从 slotToExpunge 为起点开始进行清理脏 entry 的过程。 下面用一个实例来有个直观的感受，示例代码就不给出了，代码 debug 时 table 状态如下图所示： 如图所示，当前的 staleSolt 为 i=4，首先先进行前向搜索脏 entry，当 i=3 的时候遇到脏 entry，slotToExpung 更新为 3，当 i=2 的时候 tabel 2 为 null，因此前向搜索脏 entry 的过程结束。然后进行后向环形查找，知道 i=7 的时候遇到 table 7 为 null，结束后向查找过程，并且在该过程并没有找到可以覆盖的 entry。最后只能在 staleSlot（4）处插入新 entry，然后从 slotToExpunge（3）为起点进行 cleanSomeSlots 进行脏 entry 的清理。是不是上面的 1.2 的情况。 当调用 threadLocal 的 get 方法时，当 table [i] 不是和所要找的 key 相同的话，会继续通过 threadLocalMap 的 getEntryAfterMiss 方法向后环形去找，该方法为： 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; } 当 key==null 的时候，即遇到脏 entry 也会调用 expungeStleEntry 对脏 entry 进行清理。 当调用 threadLocal.remove 方法时候，实际上会调用 threadLocalMap 的 remove 方法，该方法的源码为： private void remove(ThreadLocal<?> key) { 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)]) { if (e.get() == key) { e.clear(); expungeStaleEntry(i); return; } } } 同样的可以看出，当遇到了 key 为 null 的脏 entry 的时候，也会调用 expungeStaleEntry 清理掉脏 entry。 从以上 set,getEntry,remove 方法看出，在 threadLocal 的生命周期里，针对 threadLocal 存在的内存泄漏的问题，都会通过 expungeStaleEntry，cleanSomeSlots,replaceStaleEntry 这三个方法清理掉 key 为 null 的脏 entry。 2.4、为什么使用弱引用 通过 threadLocal,threadLocalMap,entry 的引用关系看起来 threadLocal 存在内存泄漏的问题似乎是因为 threadLocal 是被弱引用修饰的。那为什么要使用弱引用呢？ 如果使用强引用 假设 threadLocal 使用的是强引用，在业务代码中执行 threadLocalInstance==null 操作，以清理掉 threadLocal 实例的目的，但是因为 threadLocalMap 的 Entry 强引用 threadLocal，因此在 gc 的时候进行可达性分析，threadLocal 依然可达，对 threadLocal 并不会进行垃圾回收，这样就无法真正达到业务逻辑的目的，出现逻辑错误。 如果使用弱引用 假设 Entry 弱引用 threadLocal，尽管会出现内存泄漏的问题，但是在 threadLocal 的生命周期里（set,getEntry,remove）里，都会针对 key 为 null 的脏 entry 进行处理。 从以上的分析可以看出，使用弱引用的话在 threadLocal 生命周期里会尽可能的保证不出现内存泄漏的问题，达到安全的状态。 2.5、Thread.exit() 当线程退出时会执行 exit 方法： private void exit() { if (group != null) { group.threadTerminated(this); group = null; } /* Aggressively null out all reference fields: see bug 4006245 */ target = null; /* Speed the release of some of these resources */ threadLocals = null; inheritableThreadLocals = null; inheritedAccessControlContext = null; blocker = null; uncaughtExceptionHandler = null; } 从源码可以看出当线程结束时，会令 threadLocals=null，也就意味着 GC 的时候就可以将 threadLocalMap 进行垃圾回收，换句话说 threadLocalMap 生命周期实际上 thread 的生命周期相同。 三、threadLocal 最佳实践 本文对 threadLocal 的内存泄漏做了很详细的分析，我们可以完全理解 threadLocal 内存泄漏的前因后果，那么实践中我们应该怎么做？ 1、每次使用完 ThreadLocal，都调用它的 remove () 方法，清除数据。 2、在使用线程池的情况下，没有及时清理 ThreadLocal，不仅是内存泄漏的问题，更严重的是可能导致业务逻辑出现问题。所以，使用 ThreadLocal 就跟加锁完要解锁一样，用完就清理。
一、简介 在多线程编程中通常解决线程安全的问题我们会利用 synchronzed 或者 lock 控制线程对临界区资源的同步顺序从而解决线程安全的问题，但是这种加锁的方式会让未获取到锁的线程进行阻塞等待，很显然这种方式的时间效率并不是很好。线程安全问题的核心在于多个线程会对同一个临界区共享资源进行操作，那么，如果每个线程都使用自己的 共享资源，各自使用各自的，又互相不影响到彼此即让多个线程间达到隔离的状态，这样就不会出现线程安全的问题。事实上，这就是一种 “空间换时间” 的方案，每个线程都会都拥有自己的 共享资源 无疑内存会大很多，但是由于不需要同步也就减少了线程可能存在的阻塞等待的情况从而提高的时间效率。 虽然 ThreadLocal 并不在 java.util.concurrent 包中而在 java.lang 包中，但我更倾向于把它当作是一种并发容器（虽然真正存放数据的是 ThreadLocalMap）进行归类。从 ThreadLocal 这个类名可以顾名思义的进行理解，表示线程的 本地变量，即每个线程都拥有该变量副本，达到人手一份的效果，各用各的这样就可以避免共享资源的竞争。 二、实现原理 2.1、void set(T value) set 方法设置在当前线程中 threadLocal 变量的值，该方法的源码为： public void set(T value) { //1、获取当前线程实例对象 Thread t = Thread.currentThread(); //2、通过当前线程实例获取到ThreadLocalMap对象 ThreadLocalMap map = getMap(t); if (map != null) //3、如果Map不为null,则以当前threadLocl实例为key,值为value进行存入 map.set(this, value); else //4、map为null,则新建ThreadLocalMap并存入value createMap(t, value); } 首先 ThreadLocalMap 是怎样来的 ThreadLocalMap getMap(Thread t) { return t.threadLocals; } 该方法直接返回的就是当前线程对象t的一个成员变量 threadLocals： /* ThreadLocal values pertaining to this thread. This map is maintained * by the ThreadLocal class. */ ThreadLocal.ThreadLocalMap threadLocals = null; 也就是说 ThreadLocalMap 的引用是作为 Thread 的一个成员变量，被 Thread 进行维护的。回过头再来看看 set 方法，当 map 为 Null 的时候会通过 createMap (t，value) 方法： void createMap(Thread t, T firstValue) { t.threadLocals = new ThreadLocalMap(this, firstValue); } 该方法就是 new 一个 ThreadLocalMap 实例对象，然后同样以当前 threadLocal 实例作为 key, 值为 value 存放到 threadLocalMap 中，然后将当前线程对象的 threadLocals 赋值为 threadLocalMap。 通过当前线程对象 thread 获取该 thread 所维护的 threadLocalMap, 若 threadLocalMap 不为 null, 则以 threadLocal 实例为 key, 值为 value 的键值对存入 threadLocalMap, 若 threadLocalMap 为 null 的话，就新建 threadLocalMap 然后在以 threadLocal 为键，值为 value 的键值对存入即可。 2.2、T get() get 方法是获取当前线程中 threadLocal 变量的值。 public T get() { //1、获取当前线程的实例对象 Thread t = Thread.currentThread(); //2、获取当前线程的threadLocalMap ThreadLocalMap map = getMap(t); if (map != null) { //3、获取map中当前threadLocal实例为key的值的entry ThreadLocalMap.Entry e = map.getEntry(this); if (e != null) { @SuppressWarnings("unchecked") //4、当前entitiy不为null的话，就返回相应的值value T result = (T)e.value; return result; } } //5、若map为null或者entry为null的话通过该方法初始化，并返回该方法返回的value return setInitialValue(); } 弄懂了 set 方法的逻辑，看 get 方法只需要带着逆向思维去看就好，如果是那样存的，反过来去拿就好。代码逻辑请看注释，另外，看下 setInitialValue 主要做了些什么事情？ private T setInitialValue() { T value = initialValue(); Thread t = Thread.currentThread(); ThreadLocalMap map = getMap(t); if (map != null) map.set(this, value); else createMap(t, value); return value; } 这段方法的逻辑和 set 方法几乎一致，另外值得关注的是 initialValue 方法: protected T initialValue() { return null; } 这个方法是 protected 修饰的也就是说继承 ThreadLocal 的子类可重写该方法，实现赋值为其他的初始值。关于 get 方法来总结一下： 通过当前线程 thread 实例获取到它所维护的 threadLocalMap，然后以当前 threadLocal 实例为 key 获取该 map 中的键值对（Entry），若 Entry 不为 null 则返回 Entry 的 value。如果获取 threadLocalMap 为 null 或者 Entry 为 null 的话，就以当前 threadLocal 为 Key，value 为 null 存入 map 后，并返回 null。 2.3、void remove() public void remove() { //1. 获取当前线程的threadLocalMap ThreadLocalMap m = getMap(Thread.currentThread()); if (m != null) //2. 从map中删除以当前threadLocal实例为key的键值对 m.remove(this); } get,set 方法实现了存数据和读数据，当然还得学会如何删数据。删除数据当然是从 map 中删除数据，先获取与当前线程相关联的 threadLocalMap 然后从 map 中删除该 threadLocal 实例为 key 的键值对即可。 三、ThreadLocalMap 详解 数据其实都放在了 threadLocalMap 中，threadLocal 的 get，set 和 remove 方法实际上具体是通过 threadLocalMap 的 getEntry,set 和 remove 方法实现的。 3.1、Entry 数据结构 ThreadLocalMap 是 threadLocal 一个静态内部类，和大多数容器一样内部维护了一个数组，同样的 threadLocalMap 内部维护了一个 Entry 类型的 table 数组。 /** * The table, resized as necessary. * table.length MUST always be a power of two. */ private Entry[] table; 通过注释可以看出，table 数组的长度为2的幂次方。接下来看下 Entry 是什么： static class Entry extends WeakReference<ThreadLocal<?>> { /** The value associated with this ThreadLocal. */ Object value; Entry(ThreadLocal<?> k, Object v) { super(k); value = v; } } Entry 是一个以 ThreadLocal 为 key,Object 为 value 的键值对，另外需要注意的是这里的 threadLocal 是弱引用，因为 Entry 继承了 WeakReference，在 Entry 的构造方法中，调用了 super (k) 方法就会将 threadLocal 实例包装成一个 WeakReferenece。到这里可以用一个图，来理解下 thread，threadLocal，threadLocalMap，Entry 之间的关系： 注意上图中的实线表示强引用，虚线表示弱引用。如图所示，每个线程实例中可以通过 threadLocals 获取到 threadLocalMap，而 threadLocalMap 实际上就是一个以 threadLocal 实例为 key，任意对象为 value 的 Entry 数组。当我们为 threadLocal 变量赋值，实际上就是以当前 threadLocal 实例为 key，值为 value 的 Entry 往这个 threadLocalMap 中存放。需要注意的是 Entry 中的 key 是弱引用，当 threadLocal 外部强引用被置为 null (threadLocalInstance=null), 那么系统 GC 的时候，根据可达性分析，这个 threadLocal 实例就没有任何一条链路能够引用到它，这个 ThreadLocal 势必会被回收，这样一来，ThreadLocalMap 中就会出现 key 为 null 的 Entry，就没有办法访问这些 key 为 null 的 Entry 的 value，如果当前线程再迟迟不结束的话，这些 key 为 null 的 Entry 的 value 就会一直存在一条强引用链：Thread Ref -> Thread -> ThreaLocalMap -> Entry -> value 永远无法回收，造成内存泄漏。当然，如果当前 thread 运行结束，threadLocal，threadLocalMap,Entry 没有引用链可达，在垃圾回收的时候都会被系统进行回收。在实际开发中，会使用线程池去维护线程的创建和复用，比如固定大小的线程池，线程为了复用是不会主动结束的，所以，threadLocal 的内存泄漏问题，是应该值得思考和注意的问题。 3.2、set 方法 与 concurrentHashMap，hashMap 等容器一样，threadLocalMap 也是采用散列表进行实现的。 3.2.1、散列表 理想状态下，散列表就是一个包含关键字的固定大小的数组，通过使用散列函数，将关键字映射到数组的不同位置。下面是 在理想状态下，哈希函数可以将关键字均匀的分散到数组的不同位置，不会出现两个关键字散列值相同（假设关键字数量小于数组的大小）的情况。但是在实际使用中，经常会出现多个关键字散列值相同的情况（被映射到数组的同一个位置），我们将这种情况称为散列冲突。为了解决散列冲突，主要采用下面两种方式：分离链表法（separate chaining）和开放定址法（open addressing）。 3.2.2、分离链表法 分散链表法使用链表解决冲突，将散列值相同的元素都保存到一个链表中。当查询的时候，首先找到元素所在的链表，然后遍历链表查找对应的元素，典型实现为 hashMap，concurrentHashMap 的拉链法。下面是一个示意图： 3.2.3、开放定址法 开放定址法不会创建链表，当关键字散列到的数组单元已经被另外一个关键字占用的时候，就会尝试在数组中寻找其他的单元，直到找到一个空的单元。探测数组空单元的方式有很多，这里介绍一种最简单的 -- 线性探测法。线性探测法就是从冲突的数组单元开始，依次往后搜索空单元，如果到数组尾部，再从头开始搜索（环形查找）。如下图所示： 关于两种方式的比较，可以参考 这篇文章。ThreadLocalMap 中使用开放地址法来处理散列冲突，而 HashMap 中使用的分离链表法。之所以采用不同的方式主要是因为：在 ThreadLocalMap 中的散列值分散的十分均匀，很少会出现冲突。并且 ThreadLocalMap 经常需要清除无用的对象，使用纯数组更加方便。 在了解这些相关知识后我们再回过头来看一下 set 方法。set 方法的源码为： private void set(ThreadLocal<?> key, Object value) { // We don't use a fast path as with get() because it is at // least as common to use set() to create new entries as // it is to replace existing ones, in which case, a fast // path would fail more often than not. Entry[] tab = table; int len = tab.length; //根据threadLocal的hashCode确定Entry应该存放的位置 int i = key.threadLocalHashCode & (len-1); //采用开放地址法，hash冲突的时候使用线性探测 for (Entry e = tab[i]; e != null; e = tab[i = nextIndex(i, len)]) { ThreadLocal<?> k = e.get(); //覆盖旧Entry if (k == key) { e.value = value; return; } //当key为null时，说明threadLocal强引用已经被释放掉，那么就无法 //再通过这个key获取threadLocalMap中对应的entry，这里就存在内存泄漏的可能性 if (k == null) { //用当前插入的值替换掉这个key为null的“脏”entry replaceStaleEntry(key, value, i); return; } } //新建entry并插入table中i处 tab[i] = new Entry(key, value); int sz = ++size; //插入后再次清除一些key为null的“脏”entry,如果大于阈值就需要扩容 if (!cleanSomeSlots(i, sz) && sz >= threshold) rehash(); } 3.3、注意点 3.3.1、threadLocal 的 hashcode private final int threadLocalHashCode = nextHashCode(); private static final int HASH_INCREMENT = 0x61c88647; private static AtomicInteger nextHashCode =new AtomicInteger(); /** * Returns the next hash code. */ private static int nextHashCode() { return nextHashCode.getAndAdd(HASH_INCREMENT); } threadLocal 实例的 hashCode 是通过 nextHashCode () 方法实现的，该方法实际上总是用一个 AtomicInteger 加上 0x61c88647 来实现的。0x61c88647 这个数是有特殊意义的，它能够保证 hash 表的每个散列桶能够均匀的分布，这是 Fibonacci Hashing，关于更多介绍可以看 这篇文章的 threadLocal 散列值部分。也正是能够均匀分布，所以 threadLocal 选择使用开放地址法来解决 hash 冲突的问题。 3.3.2、怎样确定新值插入到哈希表中的位置 该操作源码为：key.threadLocalHashCode & (len-1)，同 hashMap 和 ConcurrentHashMap 等容器的方式一样，利用当前 key (即 threadLocal 实例) 的 hashcode 与哈希表大小相与，因为哈希表大小总是为 2 的幂次方，所以相与等同于一个取模的过程，这样就可以通过 Key 分配到具体的哈希桶中去。而至于为什么取模要通过位与运算的原因就是位运算的执行效率远远高于了取模运算。 3.3.3、怎样解决 hash 冲突？ 源码中通过 nextIndex (i, len) 方法解决 hash 冲突的问题，该方法为 ((i + 1 < len) ? i + 1 : 0);，也就是不断往后线性探测，当到哈希表末尾的时候再从 0 开始，成环形。 3.3.4、怎样解决“脏 Entry 在分析 threadLocal,threadLocalMap 以及 Entry 的关系的时候，我们已经知道使用 threadLocal 有可能存在内存泄漏（对象创建出来后，在之后的逻辑一直没有使用该对象，但是垃圾回收器无法回收这个部分的内存），在源码中针对这种 key 为 null 的 Entry 称之为 stale entry，直译为不新鲜的 entry，我把它理解为 脏 entry，自然而然，Josh Bloch and Doug Lea 大师考虑到了这种情况，在 set 方法的 for 循环中寻找和当前 Key 相同的可覆盖 entry 的过程中通过 replaceStaleEntry 方法解决脏 entry 的问题。如果当前 table [i] 为 null 的话，直接插入新 entry 后也会通过 cleanSomeSlots 来解决脏 entry 的问题。 3.3.5、如何进行扩容？ threshold 的确定。 也几乎和大多数容器一样，threadLocalMap 会有扩容机制，那么它的 threshold 又是怎样确定的了？ private int threshold; // Default to 0 /** * The initial capacity -- MUST be a power of two. */ private static final int INITIAL_CAPACITY = 16; 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); } /** * Set the resize threshold to maintain at worst a 2/3 load factor. */ private void setThreshold(int len) { threshold = len * 2 / 3; } 根据源码可知，在第一次为 threadLocal 进行赋值的时候会创建初始大小为 16 的 threadLocalMap, 并且通过 setThreshold 方法设置 threshold，其值为当前哈希数组长度乘以（2/3），也就是说加载因子为 2/3 (加载因子是衡量哈希表密集程度的一个参数，如果加载因子越大的话，说明哈希表被装载的越多，出现 hash 冲突的可能性越大，反之，则被装载的越少，出现 hash 冲突的可能性越小。同时如果过小，很显然内存使用率不高，该值取值应该考虑到内存使用率和 hash 冲突概率的一个平衡，如 hashMap,concurrentHashMap 的加载因子都为 0.75)。这里 threadLocalMap 初始大小为 16，加载因子为 2/3，所以哈希表可用大小为：16*2/3=10，即哈希表可用容量为 10。 3.3.6、扩容 resize 从 set 方法中可以看出当 hash 表的 size 大于 threshold 的时候，会通过 resize 方法进行扩容。 /** * Double the capacity of the table. */ private void resize() { Entry[] oldTab = table; int oldLen = oldTab.length; //新数组为原数组的2倍 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(); //遍历过程中如果遇到脏entry的话直接另value为null,有助于value能够被回收 if (k == null) { e.value = null; // Help the GC } else { //重新确定entry在新数组的位置，然后进行插入 int h = k.threadLocalHashCode & (newLen - 1); while (newTab[h] != null) h = nextIndex(h, newLen); newTab[h] = e; count++; } } } //设置新哈希表的threshHold和size属性 setThreshold(newLen); size = count; table = newTab; } 新建一个大小为原来数组长度的两倍的数组，然后遍历旧数组中的 entry 并将其插入到新的 hash 数组中，主要注意的是，在扩容的过程中针对脏 entry 的话会令 value 为 null，以便能够被垃圾回收器能够回收，解决隐藏的内存泄漏的问题。 3.3、getEntry 方法 getEntry 方法源码为： private Entry getEntry(ThreadLocal<?> key) { //1. 确定在散列数组中的位置 int i = key.threadLocalHashCode & (table.length - 1); //2. 根据索引i获取entry Entry e = table[i]; //3. 满足条件则返回该entry if (e != null && e.get() == key) return e; else //4. 未查找到满足条件的entry，额外在做的处理 return getEntryAfterMiss(key, i, e); } 方法逻辑很简单，若能当前定位的 entry 的 key 和查找的 key 相同的话就直接返回这个 entry，否则的话就是在 set 的时候存在 hash 冲突的情况，需要通过 getEntryAfterMiss 做进一步处理。getEntryAfterMiss 方法为： 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) //找到和查询的key相同的entry则返回 return e; if (k == null) //解决脏entry的问题 expungeStaleEntry(i); else //继续向后环形查找 i = nextIndex(i, len); e = tab[i]; } return null; } 这个方法同样很好理解，通过 nextIndex 往后环形查找，如果找到和查询的 key 相同的 entry 的话就直接返回，如果在查找过程中遇到脏 entry 的话使用 expungeStaleEntry 方法进行处理。到目前为止，为了解决潜在的内存泄漏的问题，在 set，resize,getEntry 这些地方都会对这些脏 entry 进行处理，可见为了尽可能解决这个问题几乎无时无刻都在做出努力。 3.4、remove /** * Remove the entry for key. */ private void remove(ThreadLocal<?> key) { 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)]) { if (e.get() == key) { //将entry的key置为null e.clear(); //将该entry的value也置为null expungeStaleEntry(i); return; } } } 该方法逻辑很简单，通过往后环形查找到与指定 key 相同的 entry 后，先通过 clear 方法将 key 置为 null 后，使其转换为一个脏 entry，然后调用 expungeStaleEntry 方法将其 value 置为 null，以便垃圾回收时能够清理，同时将 table [i] 置为 null。 四、ThreadLocal的使用场景 ThreadLocal 不是用来解决共享对象的多线程访问问题的，数据实质上是放在每个 thread 实例引用的 threadLocalMap, 也就是说每个不同的线程都拥有专属于自己的数据容器（threadLocalMap），彼此不影响。因此 threadLocal 只适用于共享对象会造成线程安全的业务场景。比如 hibernate 中通过 threadLocal 管理 Session 就是一个典型的案例，不同的请求线程（用户）拥有自己的 session，若将 session 共享出去被多线程访问，必然会带来线程安全问题。下面，是一个例子，SimpleDateFormat.parse 方法会有线程安全的问题，可以尝试使用 threadLocal 包装 SimpleDateFormat，将该实例不被多线程共享即可。 public class ThreadLocalDemo { private static ThreadLocal<SimpleDateFormat> sdf = new ThreadLocal<>(); public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(10); for (int i = 0; i < 100; i++) { executorService.submit(new DateUtil("2019-11-25 09:00:" + i % 60)); } } static class DateUtil implements Runnable { private String date; public DateUtil(String date) { this.date = date; } @Override public void run() { if (sdf.get() == null) { sdf.set(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")); } else { try { Date date = sdf.get().parse(this.date); System.out.println(date); } catch (ParseException e) { e.printStackTrace(); } } } } } 1、如果当前线程不持有 SimpleDateformat 对象实例，那么就新建一个并把它设置到当前线程中，如果已经持有，就直接使用。另外，从 if (sdf.get () == null){....} else {.....} 可以看出为每一个线程分配一个 SimpleDateformat 对象实例是从应用层面（业务代码逻辑）去保证的。 2、在上面说过 threadLocal 有可能存在内存泄漏，在使用完之后，最好使用 remove 方法将这个变量移除，就像在使用数据库连接一样，及时关闭连接。
一、简介 ArrayList 并不是线程安全的，在读线程在读取 ArrayList 的时候如果有写线程在写数据的时候，基于 fast-fail 机制，会抛出 ConcurrentModificationException 异常，也就是说 ArrayList 并不是一个线程安全的容器，当然您可以用 Vector, 或者使用 Collections 的静态方法将 ArrayList 包装成一个线程安全的类，但是这些方式都是采用 java 关键字 synchronzied 对方法进行修饰，利用独占式锁来保证线程安全的。但是，由于独占式锁在同一时刻只有一个线程能够获取到对象监视器，很显然这种方式效率并不是太高。 回到业务场景中，有很多业务往往是读多写少的，比如系统配置的信息，除了在初始进行系统配置的时候需要写入数据，其他大部分时刻其他模块之后对系统信息只需要进行读取，又比如白名单，黑名单等配置，只需要读取名单配置然后检测当前用户是否在该配置范围以内。类似的还有很多业务场景，它们都是属于读多写少的场景。如果在这种情况用到上述的方法，使用 Vector,Collections 转换的这些方式是不合理的，因为尽管多个读线程从同一个数据容器中读取数据，但是读线程对数据容器的数据并不会发生发生修改。很自然而然的我们会联想到 ReenTrantReadWriteLock，通过读写分离的思想，使得读读之间不会阻塞，无疑如果一个 list 能够做到被多个读线程读取的话，性能会大大提升不少。但是，如果仅仅是将 list 通过读写锁（ReentrantReadWriteLock）进行再一次封装的话，由于读写锁的特性，当写锁被写线程获取后，读写线程都会被阻塞。如果仅仅使用读写锁对 list 进行封装的话，这里仍然存在读线程在读数据的时候被阻塞的情况，如果想 list 的读效率更高的话，如果我们保证读线程无论什么时候都不被阻塞，效率岂不是会更高？ Doug Lea 大师就为我们提供 CopyOnWriteArrayList 容器可以保证线程安全，保证读读之间在任何时候都不会被阻塞，CopyOnWriteArrayList 也被广泛应用于很多业务场景之中。 二、COW 的设计思想 回到上面所说的，如果简单的使用读写锁的话，在写锁被获取之后，读写线程被阻塞，只有当写锁被释放后读线程才有机会获取到锁从而读到最新的数据，站在读线程的角度来看，即读线程任何时候都是获取到最新的数据，满足数据实时性。既然我们说到要进行优化，必然有 trade-off, 我们就可以牺牲数据实时性满足数据的最终一致性即可。而 CopyOnWriteArrayList 就是通过 Copy-On-Write (COW)，即写时复制的思想来通过延时更新的策略来实现数据的最终一致性，并且能够保证读线程间不阻塞。 COW 通俗的理解是当我们往一个容器添加元素的时候，不直接往当前容器添加，而是先将当前容器进行 Copy，复制出一个新的容器，然后新的容器里添加元素，添加完元素之后，再将原容器的引用指向新的容器。对 CopyOnWrite 容器进行并发的读的时候，不需要加锁，因为当前容器不会添加任何元素。所以 CopyOnWrite 容器也是一种读写分离的思想，延时更新的策略是通过在写的时候针对的是不同的数据容器来实现的，放弃数据实时性达到数据的最终一致性。 三、实现原理 实际上 CopyOnWriteArrayList 内部维护的就是一个数组。 /** The array, accessed only via getArray/setArray. */ private transient volatile Object[] array; 并且该数组引用是被 volatile 修饰，注意这里仅仅是修饰的是数组引用。对 list 来说，最关心的就是读写的时候，分别为 get 和 add 方法的实现。 3.1 get 方法实现原理 get 方法的源码为： public E get(int index) { return get(getArray(), index); } /** * Gets the array. Non-private so as to also be accessible * from CopyOnWriteArraySet class. */ final Object[] getArray() { return array; } private E get(Object[] a, int index) { return (E) a[index]; } 可以看出来 get 方法实现非常简单，几乎就是一个 “单线程” 程序，没有对多线程添加任何的线程安全控制，也没有加锁也没有 CAS 操作等等，原因是，所有的读线程只是会读取数据容器中的数据，并不会进行修改。 3.2、add 方法实现原理 再来看下如何进行添加数据的？add 方法的源码为： public boolean add(E e) { final ReentrantLock lock = this.lock; //1. 使用Lock,保证写线程在同一时刻只有一个 lock.lock(); try { //2. 获取旧数组引用 Object[] elements = getArray(); int len = elements.length; //3. 创建新的数组，并将旧数组的数据复制到新数组中 Object[] newElements = Arrays.copyOf(elements, len + 1); //4. 往新数组中添加新的数据 newElements[len] = e; //5. 将旧数组引用指向新的数组 setArray(newElements); return true; } finally { lock.unlock(); } } add 方法的逻辑也比较容易理解，请看上面的注释。需要注意这么几点： 1、采用 ReentrantLock，保证同一时刻只有一个写线程正在进行数组的复制，否则的话内存中会有多份被复制的数据； 2、前面说过数组引用是 volatile 修饰的，因此将旧的数组引用指向新的数组，根据 volatile 的 happens-before 规则，写线程对数组引用的修改对读线程是可见的。 3、由于在写数据的时候，是在新的数组中插入数据的，从而保证读写实在两个不同的数据容器中进行操作。 四、总结 COW和读写锁都是通过读写分离的思想实现的，但两者还是有些不同，可以进行比较： 4.1、COW vs 读写锁 相同点： 1、两者都是通过读写分离的思想实现； 2、读线程间是互不阻塞的 不同点：对读线程而言，为了实现数据实时性，在写锁被获取后，读线程会等待或者当读锁被获取后，写线程会等待，从而解决 “脏读” 等问题。也就是说如果使用读写锁依然会出现读线程阻塞等待的情况。而 COW 则完全放开了牺牲数据实时性而保证数据最终一致性，即读线程对数据的更新是延时感知的，因此读线程不会存在等待的情况。 对这一点从文字上还是很难理解，通过 debug 看一下，add 方法核心代码为： Object[] elements = getArray(); int len = elements.length; Object[] newElements = Arrays.copyOf(elements, len + 1); newElements[len] = e; setArray(newElements); 假设 COW 的变化如下图所示： 数组中已有数据 1,2,3，现在写线程想往数组中添加数据 4，我们在第 5 行处打上断点，让写线程暂停。读线程依然会 “不受影响” 的能从数组中读取数据，可是还是只能读到 1,2,3。如果读线程能够立即读到新添加的数据的话就叫做能保证数据实时性 。当对第 5 行的断点放开后，读线程才能感知到数据变化，读到完整的数据 1,2,3,4，而保证数据最终一致性，尽管有可能中间间隔了好几秒才感知到。 这里还有这样一个问题： 为什么需要复制呢？ 如果将 array 数组设定为 volitile 的， 对 volatile 变量写 happens-before 读，读线程不是能够感知到 volatile 变量的变化。 原因是，这里 volatile 的修饰的仅仅只是数组引用，数组中的元素的修改是不能保证可见性的。因此 COW 采用的是新旧两个数据容器，通过第 5 行代码将数组引用指向新的数组。 这也是为什么 concurrentHashMap 只具有弱一致性的原因，关于 concurrentHashMap 的弱一致性可以。 4.2、COW 的缺点 CopyOnWrite 容器有很多优点，但是同时也存在两个问题，即内存占用问题和数据一致性问题。所以在开发的时候需要注意一下。 1、内存占用问题：因为 CopyOnWrite 的写时复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对象的内存，旧的对象和新写入的对象（注意：在复制的时候只是复制容器里的引用，只是在写的时候会创建新对象添加到新容器里，而旧容器的对象还在使用，所以有两份对象内存）。如果这些对象占用的内存比较大，比如说 200M 左右，那么再写入 100M 数据进去，内存就会占用 300M，那么这个时候很有可能造成频繁的 minor GC 和 major GC。 2、数据一致性问题：CopyOnWrite 容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果你希望写入的的数据，马上能读到，请不要使用 CopyOnWrite 容器。
一、简介 在单线程编程中会经常用到一些集合类，比如 ArrayList,HashMap 等，但是这些类都不是线程安全的类。在面试中也经常会有一些考点，比如 ArrayList 不是线程安全的，Vector 是线程安全。而保障 Vector 线程安全的方式，是非常粗暴的在方法上用 synchronized 独占锁，将多线程执行变成串行化。要想将 ArrayList 变成线程安全的也可以使用 Collections.synchronizedList (List<T> list) 方法 ArrayList 转换成线程安全的，但这种转换方式依然是通过 synchronized 修饰方法实现的，很显然这不是一种高效的方式，同时，队列也是我们常用的一种数据结构，为了解决线程安全的问题，Doug Lea 大师为我们准备了 ConcurrentLinkedQueue 这个线程安全的队列。从类名就可以看的出来实现队列的数据结构是链式。 1.1、Node 学习 ConcurrentLinkedQueue 自然而然得先从它的节点类看起，明白它的底层数据结构。Node 类的源码为： private static class Node<E> { volatile E item; volatile Node<E> next; ....... } Node 节点主要包含了两个域：一个是数据域 item，另一个是 next 指针，用于指向下一个节点从而构成链式队列。并且都是用 volatile 进行修饰的，以保证内存可见性。另外 ConcurrentLinkedQueue 含有这样两个成员变量： private transient volatile Node<E> head; private transient volatile Node<E> tail; 说明 ConcurrentLinkedQueue 通过持有头尾指针进行管理队列。当调用无参构造器时，其源码为： public ConcurrentLinkedQueue() { head = tail = new Node<E>(null); } head 和 tail 指针会指向一个 item 域为 null 的节点，此时 ConcurrentLinkedQueue 状态如下图所示： 如图，head 和 tail 指向同一个节点 Node0，该节点 item 域为 null,next 域为 null。 1.2 操作 Node 的几个 CAS 操作 在队列进行出队入队的时候免不了对节点需要进行操作，在多线程就很容易出现线程安全的问题。可以看出在处理器指令集能够支持 CMPXCHG 指令后，在 java 源码中涉及到并发处理都会使用 CAS 操作，那么在 ConcurrentLinkedQueue 对 Node 的 CAS 操作有这样几个： //更改Node中的数据域item boolean casItem(E cmp, E val) { return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val); } //更改Node中的指针域next void lazySetNext(Node<E> val) { UNSAFE.putOrderedObject(this, nextOffset, val); } //更改Node中的指针域next boolean casNext(Node<E> cmp, Node<E> val) { return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val); } 可以看出这些方法实际上是通过调用 UNSAFE 实例的方法，UNSAFE 为 sun.misc.Unsafe 类，该类是 hotspot 底层方法，目前为止了解即可，知道 CAS 的操作归根结底是由该类提供就好。 二、offer 方法 对一个队列来说，插入满足 FIFO 特性，插入元素总是在队列最末尾的地方进行插入，而取（移除）元素总是从队列的队头。所有要想能够彻底弄懂 ConcurrentLinkedQueue 自然而然是从 offer 方法和 poll 方法开始。那么为了能够理解 offer 方法，采用 debug 的方式来一行一行的看代码走。另外，在看多线程的代码时，可采用这样的思维方式： 单个线程 offer 多个线程 offer 部分线程 offer，部分线程 poll offer 的速度快于 poll 队列长度会越来越长，由于 offer 节点总是在对队列队尾，而 poll 节点总是在队列对头，也就是说 offer 线程和 poll 线程两者并无 “交集”，也就是说两类线程间并不会相互影响，这种情况站在相对速率的角度来看，也就是一个 "单线程 offer" offer 的速度慢于 poll poll 的相对速率快于 offer，也就是队头删的速度要快于队尾添加节点的速度，导致的结果就是队列长度会越来越短，而 offer 线程和 poll 线程就会出现 “交集”，即那一时刻就可以称之为 offer 线程和 poll 线程同时操作的节点为 ** 临界点 ** ，且在该节点 offer 线程和 poll 线程必定相互影响。根据在临界点时 offer 和 poll 发生的相对顺序又可从两个角度去思考：1、执行顺序为 offer-->poll-->offer，即表现为当 offer 线程在 Node1 后插入 Node2 时，此时 poll 线程已经将 Node1 删除，这种情况很显然需要在 offer 方法中考虑；2、执行顺序可能为：poll-->offer-->poll，即表现为当 poll 线程准备删除的节点为 null 时（队列为空队列），此时 offer 线程插入一个节点使得队列变为非空队列。 先看这么一段代码： 1. ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>(); 2. queue.offer(1); 3. queue.offer(2); 创建一个 ConcurrentLinkedQueue 实例，先 offer 1，然后再 offer 2。offer 的源码为： public boolean offer(E e) { checkNotNull(e); final Node<E> newNode = new Node<E>(e); for (Node<E> t = tail, p = t;;) { Node<E> q = p.next; if (q == null) { // p is last node if (p.casNext(null, newNode)) { // Successful CAS is the linearization point // for e to become an element of this queue, // and for newNode to become "live". if (p != t) // hop two nodes at a time casTail(t, newNode); // Failure is OK. return true; } // Lost CAS race to another thread; re-read next } else if (p == q) // We have fallen off list. If tail is unchanged, it // will also be off-list, in which case we need to // jump to head, from which all live nodes are always // reachable. Else the new tail is a better bet. p = (t != (t = tail)) ? t : head; else // Check for tail updates after two hops. p = (p != t && t != (t = tail)) ? t : q; } } 2.1、单线程执行角度分析 分析 offer 1 的过程。第 1 行代码会对是否为 null 进行判断，为 null 的话就直接抛出空指针异常，第 2 行代码将 e 包装成一个 Node 类，第 3 行为 for 循环，只有初始化条件没有循环结束条件，这很符合 CAS 的 “套路”，在循环体 CAS 操作成功会直接 return 返回，如果 CAS 操作失败的话就在 for 循环中不断重试直至成功。这里实例变量 t 被初始化为 tail，p 被初始化为 t 即 tail。为了方便下面的理解，**p 被认为队列真正的尾节点，tail 不一定指向对象真正的尾节点，因为在 ConcurrentLinkedQueue 中 tail 是被延迟更新的 **，具体原因我们慢慢来看。代码走到第 3 行的时候，t 和 p 都分别指向初始化时创建的 item 域为 null，next 域为 null 的 Node0。第 4 行变量 q 被赋值为 null，第 5 行 if 判断为 true，在第 7 行使用 casNext 将插入的 Node 设置成当前队列尾节点 p 的 next 节点，如果 CAS 操作失败，此次循环结束在下次循环中进行重试。CAS 操作成功走到第 8 行，此时 p==t，if 判断为 false, 直接 return true 返回。如果成功插入 1 的话，此时 ConcurrentLinkedQueue 的状态如下图所示： 如图，此时队列的尾节点应该为 Node1, 而 tail 指向的节点依然还是 Node0, 因此可以说明 tail 是延迟更新的。那么我们继续来看 offer 2 的时候的情况，很显然此时第 4 行 q 指向的节点不为 null 了，而是指向 Node1, 第 5 行 if 判断为 false, 第 11 行 if 判断为 false, 代码会走到第 13 行。好了，再插入节点的时候我们会问自己这样一个问题？上面已经解释了 tail 并不是指向队列真正的尾节点，那么在插入节点的时候，我们是不是应该最开始做的就是找到队列当前的尾节点在哪里才能插入？那么第 13 行代码就是找出队列真正的尾节点。 定位队列真正的对尾节点： p = (p != t && t != (t = tail)) ? t : q; 如果这段代码在 单线程环境 执行时，很显然由于 p==t, 此时 p 会被赋值为 q, 而 q 等于 Node<E> q = p.next，即 Node1。在第一次循环中指针 p 指向了队列真正的队尾节点 Node1，那么在下一次循环中第 4 行 q 指向的节点为 null，那么在第 5 行中 if 判断为 true, 那么在第 7 行依然通过 casNext 方法设置 p 节点的 next 为当前新增的 Node, 接下来走到第 8 行，这个时候 p!=t，第 8 行 if 判断为 true, 会通过 casTail (t, newNode) 将当前节点 Node 设置为队列的队尾节点，此时的队列状态示意图如下图所示： tail 指向的节点由 Node0 改变为 Node2, 这里的 casTail 失败不需要重试的原因是，offer 代码中主要是通过 p 的 next 节点 q (Node<E> q = p.next) 决定后面的逻辑走向的，当 casTail 失败时状态示意图如下： 如图，如果这里 casTail 设置 tail 失败即 tail 还是指向 Node0 节点的话，无非就是多循环几次通过 13 行代码定位到队尾节点。 通过对单线程执行角度进行分析，可以了解到 poll 的执行逻辑为： 1、如果 tail 指向的节点的下一个节点（next 域）为 null 的话，说明 tail 指向的节点即为队列真正的队尾节点，因此可以通过 casNext 插入当前待插入的节点，但此时 tail 并未变化，如图 2; 2、如果 tail 指向的节点的下一个节点（next 域）不为 null 的话，说明 tail 指向的节点不是队列的真正队尾节点。通过 q（Node<E> q = p.next） 指针往前递进去找到队尾节点，然后通过 casNext 插入当前待插入的节点，并通过 casTail 方式更改 tail，如图 3。 我们回过头再来看 p = (p != t && t != (t = tail)) ? t : q; 这行代码在单线程中，这段代码永远不会将 p 赋值为 t, 那么这么写就不会有任何作用，那我们试着在 ** 多线程 ** 的情况下进行分析。 2.3、多线程执行角度分析 多个线程 offer 很显然这么写另有深意，其实在 ** 多线程环境 ** 下这行代码很有意思的。 t != (t = tail) 这个操作并非一个原子操作，有这样一种情况： 如图，假设线程 A 此时读取了变量 t，线程 B 刚好在这个时候 offer 一个 Node 后，此时会修改 tail 指针，那么这个时候线程 A 再次执行 t=tail 时 t 会指向另外一个节点，很显然线程 A 前后两次读取的变量 t 指向的节点不相同，即 t != (t = tail) 为 true, 并且由于 t 指向节点的变化 p != t 也为 true，此时该行代码的执行结果为 p 和 t 最新的 t 指针指向了同一个节点，并且此时 t 也是队列真正的对尾节点。那么，现在已经定位到队列真正的队尾节点，就可以执行 offer 操作了。 offer->poll->offer 那么还剩下第 11 行的代码我们没有分析，大致可以猜想到应该就是回答 ** 一部分线程 offer，一部分 poll** 的这种情况。当 if (p == q) 为 true 时，说明 p 指向的节点的 next 也指向它自己，这种节点称之为 ** 哨兵节点 ， 这种节点在队列中存在的价值不大，一般表示为要删除的节点或者是空节点 **。为了能够很好的理解这种情况，我们先看看 poll 方法的执行过程后，再回过头来看，总之这是一个很有意思的事情 :)。 三、poll 方法 poll 方法源码如下： public E poll() { restartFromHead: for (;;) { for (Node<E> h = head, p = h, q;;) { E item = p.item; if (item != null && p.casItem(item, null)) { // Successful CAS is the linearization point // for item to be removed from this queue. if (p != h) // hop two nodes at a time updateHead(h, ((q = p.next) != null) ? q : p); return item; } else if ((q = p.next) == null) { updateHead(h, p); return null; } else if (p == q) continue restartFromHead; else p = q; } } } 3.1、单线程执行情况分析 还是先站在 ** 单线程的角度 ** 去理清该方法的基本逻辑。假设 ConcurrentLinkedQueue 初始状态如下图所示： 参数 offer 时的定义，我们还是先将变量 p 作为队列要删除真正的队头节点，h（head）指向的节点并不一定是队列的队头节点。先来看 poll 出 Node1 时的情况，由于 p=h=head，参照上图，很显然此时 p 指向的 Node1 的数据域不为 null, 在第 4 行代码中 item!=null 判断为 true 后接下来通过 casItem 将 Node1 的数据域设置为 null。如果 CAS 设置失败则此次循环结束等待下一次循环进行重试。若第 4 行执行成功进入到第 5 行代码，此时 p 和 h 都指向 Node1, 第 5 行 if 判断为 false, 然后直接到第 7 行 return 回 Node1 的数据域 1，方法运行结束，此时的队列状态如下图。 下面继续从队列中 poll，很显然当前 h 和 p 指向的 Node1 的数据域为 null，那么第一件事就是要 ** 定位准备删除的队头节点 (找到数据域不为 null 的节点)。 定位删除的队头节点 继续看，第三行代码 item 为 null, 第 4 行代码 if 判断为 false, 走到第 8 行代码（q = p.next）if 也为 false，由于 q 指向了 Node2, 在第 11 行的 if 判断也为 false，因此代码走到了第 13 行，这个时候 p 和 q 共同指向了 Node2, 也就找到了要删除的真正的队头节点。可以总结出，定位待删除的队头节点的过程为：** 如果当前节点的数据域为 null，很显然该节点不是待删除的节点，就用当前节点的下一个节点去试探 **。在经过第一次循环后，此时状态图为下图： 进行下一次循环，第 4 行的操作同上述，当前假设第 4 行中 casItem 设置成功，由于 p 已经指向了 Node2, 而 h 还依旧指向 Node1, 此时第 5 行的 if 判断为 true，然后执行 updateHead (h, ((q = p.next) != null) ? q : p)，此时 q 指向的 Node3，所有传入 updateHead 方法的分别是指向 Node1 的 h 引用和指向 Node3 的 q 引用。updateHead 方法的源码为： final void updateHead(Node<E> h, Node<E> p) { if (h != p && casHead(h, p)) h.lazySetNext(h); } 该方法主要是通过 casHead 将队列的 head 指向 Node3, 并且通过 h.lazySetNext 将 Node1 的 next 域指向它自己。最后在第 7 行代码中返回 Node2 的值。此时队列的状态如下图所示： Node1 的 next 域指向它自己，head 指向了 Node3。如果队列为空队列的话，就会执行到代码的第 8 行 (q = p.next) == null，if 判断为 true, 因此在第 10 行中直接返回 null。以上的分析是从单线程执行的角度去看，也可以让我们了解 poll 的整体思路，现在来做一个总结： 1、如果当前 head,h 和 p 指向的节点的 Item 不为 null 的话，说明该节点即为真正的队头节点（待删除节点），只需要通过 casItem 方法将 item 域设置为 null, 然后将原来的 item 直接返回即可。 2、如果当前 head,h 和 p 指向的节点的 item 为 null 的话，则说明该节点不是真正的待删除节点，那么应该做的就是寻找 item 不为 null 的节点。通过让 q 指向 p 的下一个节点（q = p.next）进行试探，若找到则通过 updateHead 方法更新 head 指向的节点以及构造哨兵节点（通过 updateHead 方法的 h.lazySetNext (h)）。 接下来，按照上面分析 offer 的思维方式，下面来分析一下多线程的情况，第一种情况是； 3.2、多线程执行情况分析 多个线程 poll 现在回过头来看 poll 方法的源码，有这样一部分： else if (p == q) continue restartFromHead; 这一部分就是处理多个线程 poll 的情况，q = p.next 也就是说 q 永远指向的是 p 的下一个节点，那么什么情况下会使得 p,q 指向同一个节点呢？根据上面我们的分析，只有 p 指向的节点在 poll 的时候转变成了哨兵节点（通过 updateHead 方法中的 h.lazySetNext）。当线程 A 在判断 p==q 时，线程 B 已经将执行完 poll 方法将 p 指向的节点转换为 ** 哨兵节点 ** 并且 head 指向的节点已经发生了改变，所以就需要从 restartFromHead 处执行，保证用到的是最新的 head。 poll->offer->poll 试想，还有这样一种情况，如果当前队列为空队列，线程 A 进行 poll 操作，同时线程 B 执行 offer，然后线程 A 在执行 poll，那么此时线程 A 返回的是 null 还是线程 B 刚插入的最新的那个节点呢？我们来写一代 demo： public static void main(String[] args) { Thread thread1 = new Thread(() -> { Integer value = queue.poll(); System.out.println(Thread.currentThread().getName() + " poll 的值为：" + value); System.out.println("queue当前是否为空队列：" + queue.isEmpty()); }); thread1.start(); Thread thread2 = new Thread(() -> { queue.offer(1); }); thread2.start(); } 输出结果为： Thread-0 poll 的值为：null queue当前是否为空队列：false 通过 debug 控制线程 thread1 和线程 thread2 的执行顺序，thread1 先执行到第 8 行代码 if ((q = p.next) == null)，由于此时队列为空队列 if 判断为 true，进入 if 块，此时先让 thread1 暂停，然后 thread2 进行 offer 插入值为 1 的节点后，thread2 执行结束。再让 thread1 执行，这时 **thread1 并没有进行重试 **，而是代码继续往下走，返回 null，尽管此时队列由于 thread2 已经插入了值为 1 的新的节点。所以输出结果为 thread0 poll 的为 null, 然队列不为空队列。因此，在判断队列是否为空队列的时候是不能通过线程在 poll 的时候返回为 null 进行判断的，可以通过 isEmpty 方法进行判断。 四、offer 方法中部分线程 offer 部分线程 poll 在分析 offer 方法的时候我们还留下了一个问题，即对 offer 方法中第 11 行代码的理解。 offer->poll->offer 在 offer 方法的第 11 行代码 if (p == q)，能够让 if 判断为 true 的情况为 p 指向的节点为 ** 哨兵节点 ，而什么时候会构造哨兵节点呢？在对 poll 方法的讨论中，我们已经找到了答案，即 ** 当 head 指向的节点的 item 域为 null 时会寻找真正的队头节点，等到待插入的节点插入之后，会更新 head，并且将原来 head 指向的节点设置为哨兵节点。 假设队列初始状态如下图所示： 因此在线程 A 执行 offer 时，线程 B 执行 poll 就会存在如下一种情况： 如图，线程 A 的 tail 节点存在 next 节点 Node1, 因此会通过引用 q 往前寻找队列真正的队尾节点，当执行到判断 if (p == q) 时，此时线程 B 执行 poll 操作，在对线程 B 来说，head 和 p 指向 Node0, 由于 Node0 的 item 域为 null, 同样会往前递进找到队列真正的队头节点 Node1, 在线程 B 执行完 poll 之后，Node0 就会转换为 ** 哨兵节点 **，也就意味着队列的 head 发生了改变，此时队列状态为下图。 此时线程 A 在执行判断 if (p == q) 时就为 true, 会继续执行 p = (t != (t = tail)) ? t : head;，由于 tail 指针没有发生改变所以 p 被赋值为 head, 重新从 head 开始完成插入操作。 五、HOPS 的设计 通过上面对 offer 和 poll 方法的分析，我们发现 tail 和 head 是延迟更新的，两者更新触发时机为： tail 更新触发时机：当 tail 指向的节点的下一个节点不为 null 的时候，会执行定位队列真正的队尾节点的操作，找到队尾节点后完成插入之后才会通过 casTail 进行 tail 更新；当 tail 指向的节点的下一个节点为 null 的时候，只插入节点不更新 tail。 head 更新触发时机：当 head 指向的节点的 item 域为 null 的时候，会执行定位队列真正的队头节点的操作，找到队头节点后完成删除之后才会通过 updateHead 进行 head 更新；当 head 指向的节点的 item 域不为 null 的时候，只删除节点不更新 head。 并且在更新操作时，源码中会有注释为：hop two nodes at a time。所以这种延迟更新的策略就被叫做 HOPS 的大概原因是这个（猜的 :)），从上面更新时的状态图可以看出，head 和 tail 的更新是 “跳着的” 即中间总是间隔了一个。那么这样设计的意图是什么呢？ 如果让 tail 永远作为队列的队尾节点，实现的代码量会更少，而且逻辑更易懂。但是，这样做有一个缺点，** 如果大量的入队操作，每次都要执行 CAS 进行 tail 的更新，汇总起来对性能也会是大大的损耗。如果能减少 CAS 更新的操作，无疑可以大大提升入队的操作效率，所以 doug lea 大师每间隔 1 次（tail 和队尾节点的距离为 1）进行才利用 CAS 更新 tail。** 对 head 的更新也是同样的道理，虽然，这样设计会多出在循环中定位队尾节点，但总体来说读的操作效率要远远高于写的性能，因此，多出来的在循环中定位尾节点的操作的性能损耗相对而言是很小的。
一、简介 在使用 HashMap 时在多线程情况下扩容会出现 CPU 接近 100% 的情况，因为 hashmap 并不是线程安全的，通常可以使用在 java 体系中古老的 hashtable 类，该类基本上所有的方法都采用 synchronized 进行线程安全的控制，可想而知，在高并发的情况下，每次只有一个线程能够获取对象监视器锁，这样的并发性能的确不令人满意。另外一种方式通过 Collections 的 Map<K,V> synchronizedMap (Map<K,V> m) 将 hashmap 包装成一个线程安全的 map。比如 SynchronzedMap 的 put 方法源码为： public V put(K key, V value) { synchronized (mutex) {return m.put(key, value);} } 实际上 SynchronizedMap 实现依然是采用 synchronized 独占式锁进行线程安全的并发控制的。同样，这种方案的性能也是令人不太满意的。针对这种境况，Doug Lea 大师不遗余力的为我们创造了一些线程安全的并发容器，让每一个 java 开发人员倍感幸福。相对于 hashmap 来说，ConcurrentHashMap 就是线程安全的 map，其中利用了锁分段的思想提高了并发度。 ConcurrentHashMap 在 JDK1.6 的版本网上资料很多，有兴趣的可以去看看。 JDK 1.6 版本关键要素： 1、segment 继承了 ReentrantLock 充当锁的角色，为每一个 segment 提供了线程安全的保障； 2、segment 维护了哈希散列表的若干个桶，每个桶由 HashEntry 构成的链表。 而到了 JDK 1.8 的 ConcurrentHashMap 就有了很大的变化，光是代码量就足足增加了很多。1.8 版本舍弃了 segment，并且大量使用了 synchronized，以及 CAS 无锁操作以保证 ConcurrentHashMap 操作的线程安全性。至于为什么不用 ReentrantLock 而是 Synchronzied 呢？实际上，synchronzied 做了很多的优化，包括偏向锁，轻量级锁，重量级锁，可以依次向上升级锁状态，但不能降级，因此，使用 synchronized 相较于 ReentrantLock 的性能会持平甚至在某些情况更优，具体的性能测试可以去网上查阅一些资料。另外，底层数据结构改变为采用数组 + 链表 + 红黑树的数据形式。 二、关键属性及类 在了解 ConcurrentHashMap 的具体方法实现前，需要系统的来看一下几个关键的地方。 2.1、ConcurrentHashMap 的关键属性 2.1.1、table volatile Node<K,V>[] table:// 装载 Node 的数组，作为 ConcurrentHashMap 的数据容器，采用懒加载的方式，直到第一次插入数据的时候才会进行初始化操作，数组的大小总是为 2 的幂次方。 2.1.2、nextTable volatile Node<K,V>[] nextTable; // 扩容时使用，平时为 null，只有在扩容的时候才为非 null 2.1.3、sizeCtl volatile int sizeCtl; 该属性用来控制 table 数组的大小，根据是否初始化和是否正在扩容有几种情况： 当值为负数时：如果为 - 1 表示正在初始化，如果为 - N 则表示当前正有 N-1 个线程进行扩容操作； 当值为正数时：如果当前数组为 null 的话表示 table 在初始化过程中，sizeCtl 表示为需要新建数组的长度； 若已经初始化了，表示当前数据容器（table 数组）可用容量也可以理解成临界值（插入节点数超过了该临界值就需要扩容），具体指为数组的长度 n 乘以 加载因子 loadFactor； 当值为 0 时，即数组长度为默认初始值。 2.1.3、sun.misc.Unsafe U 在 ConcurrentHashMapde 的实现中可以看到大量的 U.compareAndSwapXXXX 的方法去修改 ConcurrentHashMap 的一些属性。这些方法实际上是利用了 CAS 算法保证了线程安全性，这是一种乐观策略，假设每一次操作都不会产生冲突，当且仅当冲突发生的时候再去尝试。而 CAS 操作依赖于现代处理器指令集，通过底层 CMPXCHG 指令实现。CAS (V,O,N) 核心思想为：若当前变量实际值 V 与期望的旧值 O 相同，则表明该变量没被其他线程进行修改，因此可以安全的将新值 N 赋值给变量；若当前变量实际值 V 与期望的旧值 O 不相同，则表明该变量已经被其他线程做了处理，此时将新值 N 赋给变量操作就是不安全的，在进行重试。而在大量的同步组件和并发容器的实现中使用 CAS 是通过 sun.misc.Unsafe 类实现的，该类提供了一些可以直接操控内存和线程的底层操作，可以理解为 java 中的 “指针”。该成员变量的获取是在静态代码块中： static { try { U = sun.misc.Unsafe.getUnsafe(); ....... } catch (Exception e) { throw new Error(e); } } 2.2、ConcurrentHashMap 中关键内部类 2.2.1、Node Node 类实现了 Map.Entry 接口，主要存放 key-value 对，并且具有 next 域。 static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; volatile Node<K,V> next; ...... } 另外可以看出很多属性都是用 volatile 进行修饰的，也就是为了保证内存可见性。 2.2.2、TreeNode 树节点，继承于承载数据的 Node 类。而红黑树的操作是针对 TreeBin 类的，从该类的注释也可以看出，也就是 TreeBin 会将 TreeNode 进行再一次封装。 ** * Nodes for use in TreeBins */ static final class TreeNode<K,V> extends Node<K,V> { TreeNode<K,V> parent; // red-black tree links TreeNode<K,V> left; TreeNode<K,V> right; TreeNode<K,V> prev; // needed to unlink next upon deletion boolean red; ...... } 2.2.3、TreeBin 这个类并不负责包装用户的 key、value 信息，而是包装的很多 TreeNode 节点。实际的 ConcurrentHashMap “数组” 中，存放的是 TreeBin 对象，而不是 TreeNode 对象。 static final class TreeBin<K,V> extends Node<K,V> { TreeNode<K,V> root; volatile TreeNode<K,V> first; volatile Thread waiter; volatile int lockState; // values for lockState static final int WRITER = 1; // set while holding write lock static final int WAITER = 2; // set when waiting for write lock static final int READER = 4; // increment value for setting read lock ...... } 2.2.4、ForwardingNode 在扩容时才会出现的特殊节点，其 key,value,hash 全部为 null。并拥有 nextTable 指针引用新的 table 数组。 static final class ForwardingNode<K,V> extends Node<K,V> { final Node<K,V>[] nextTable; ForwardingNode(Node<K,V>[] tab) { super(MOVED, null, null, null); this.nextTable = tab; } ..... } 2.3、CAS 关键操作 在上面提及到在 ConcurrentHashMap 中会大量使用 CAS 修改它的属性和一些操作。因此，在理解 ConcurrentHashMap 的方法前我们需要了解下面几个常用的利用 CAS 算法来保障线程安全的操作。 2.3.1、tabAt static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); } 该方法用来获取 table 数组中索引为 i 的 Node 元素。 2.3.2、casTabAt static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); } 利用 CAS 操作设置 table 数组中索引为 i 的元素 2.3.3、setTabAt static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); } 该方法用来设置 table 数组中索引为i的元素. 三、重点方法讲解 依次看看几个常用的方法是怎样实现的。 3.1、实例构造器方法 在使用 ConcurrentHashMap 第一件事自然而然就是 new 出来一个 ConcurrentHashMap 对象，一共提供了如下几个构造器方法： // 1. 构造一个空的 map，即 table 数组还未初始化，初始化放在第一次插入数据时，默认大小为 16 ConcurrentHashMap () // 2. 给定 map 的大小 ConcurrentHashMap (int initialCapacity) // 3. 给定一个 map ConcurrentHashMap (Map<? extends K, ? extends V> m) // 4. 给定 map 的大小以及加载因子 ConcurrentHashMap (int initialCapacity, float loadFactor) // 5. 给定 map 大小，加载因子以及并发度（预计同时操作数据的线程） ConcurrentHashMap (int initialCapacity,float loadFactor, int concurrencyLevel) ConcurrentHashMap 一共给我们提供了 5 中构造器方法，具体使用请看注释，来看看第 2 种构造器，传入指定大小时的情况，该构造器源码为： public ConcurrentHashMap(int initialCapacity) { //1. 小于0直接抛异常 if (initialCapacity < 0) throw new IllegalArgumentException(); //2. 判断是否超过了允许的最大值，超过了话则取最大值，否则再对该值进一步处理 int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); //3. 赋值给sizeCtl this.sizeCtl = cap; } 这段代码的逻辑请看注释，很容易理解，如果小于 0 就直接抛出异常，如果指定值大于了所允许的最大值的话就取最大值，否则，在对指定值做进一步处理。最后将 cap 赋值给 sizeCtl, 关于 sizeCtl 的说明请看上面的说明，当调用构造器方法之后，sizeCtl 的大小应该就代表了 ConcurrentHashMap 的大小，即 table 数组长度。tableSizeFor 做了哪些事情了？源码为： /** * Returns a power of two table size for the given desired capacity. * See Hackers Delight, sec 3.2 */ private static final int tableSizeFor(int c) { int n = c - 1; n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16; return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1; } 该方法会将调用构造器方法时指定的大小转换成一个 2 的幂次方数，也就是说 ConcurrentHashMap 的大小一定是 2 的幂次方，比如，当指定大小为 18 时，为了满足 2 的幂次方特性，实际上 concurrentHashMapd 的大小为 2 的 5 次方（32）。另外，需要注意的是，调用构造器方法的时候并未构造出 table 数组（可以理解为 ConcurrentHashMap 的数据容器），只是算出 table 数组的长度，当第一次向 ConcurrentHashMap 插入数据的时候才真正的完成初始化创建 table 数组的工作。 3.2、initTable 方法 源码 private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) // 1. 保证只有一个线程正在进行初始化操作 Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { // 2. 得出数组的大小 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") // 3. 这里才真正的初始化数组 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = tab = nt; // 4. 计算数组中可用的大小：实际大小n*0.75（加载因子） sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; } 代码的逻辑请见注释，有可能存在一个情况是多个线程同时走到这个方法中，为了保证能够正确初始化，在第 1 步中会先通过 if 进行判断，若当前已经有一个线程正在初始化即 sizeCtl 值变为 - 1，这个时候其他线程在 If 判断为 true 从而调用 Thread.yield () 让出 CPU 时间片。正在进行初始化的线程会调用 U.compareAndSwapInt 方法将 sizeCtl 改为 - 1 即正在初始化的状态。另外还需要注意的事情是，在第四步中会进一步计算数组中可用的大小即为数组实际大小 n 乘以加载因子 0.75. 可以看看这里乘以 0.75 是怎么算的，0.75 为四分之三，这里 n - (n>>> 2) 是不是刚好是 n-(1/4) n=(3/4) n，挺有意思的吧：)。如果选择是无参的构造器的话，这里在 new Node 数组的时候会使用默认大小为 DEFAULT_CAPACITY（16），然后乘以加载因子 0.75 为 12，也就是说数组的可用大小为 12。 3.3、put 方法 使用 ConcurrentHashMap 最长用的也应该是 put 和 get 方法了吧，我们先来看看 put 方法是怎样实现的。调用 put 方法时实际具体实现是 putVal 方法，源码如下： /** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); //1. 计算key的hash值 int hash = spread(key.hashCode()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //2. 如果当前table还没有初始化先调用initTable方法将tab进行初始化 if (tab == null || (n = tab.length) == 0) tab = initTable(); //3. tab中索引为i的位置的元素为null，则直接使用CAS将值插入即可 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } //4. 当前正在扩容 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; synchronized (f) { if (tabAt(tab, i) == f) { //5. 当前为链表，在链表中插入新的键值对 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; } } } // 6.当前为红黑树，将新的键值对插入到红黑树中 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; } } } } // 7.插入完键值对后再根据实际大小看是否需要转换成红黑树 if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } //8.对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容 addCount(1L, binCount); return null; } put 方法的代码量有点长，我们按照上面的分解的步骤一步步来看。从整体而言，为了解决线程安全的问题，ConcurrentHashMap 使用了 synchronzied 和 CAS 的方式。在之前了解过 HashMap 以及 1.8 版本之前的 ConcurrenHashMap 都应该知道 ConcurrentHashMap 结构图，为了方面下面的讲解这里先直接给出。 如图（图片摘自网络），ConcurrentHashMap 是一个哈希桶数组，如果不出现哈希冲突的时候，每个元素均匀的分布在哈希桶数组中。当出现哈希冲突的时候，是标准的链地址的解决方式，将 hash 值相同的节点构成链表的形式，称为 “拉链法”，另外，在 1.8 版本中为了防止拉链过长，当链表的长度大于 8 的时候会将链表转换成红黑树。table 数组中的每个元素实际上是单链表的头结点或者红黑树的根节点。当插入键值对时首先应该定位到要插入的桶，即插入 table 数组的索引 i 处。那么，怎样计算得出索引 i 呢？当然是根据 key 的 hashCode 值。 1、spread () 重哈希，以减小 Hash 冲突 对于一个 hash 表来说，hash 值分散的不够均匀的话会大大增加哈希冲突的概率，从而影响到 hash 表的性能。因此通过 spread 方法进行了一次重 hash 从而大大减小哈希冲突的可能性。spread 方法为： static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS; } 该方法主要是将 key 的 hashCode 的低 16 位于高 16 位进行异或运算，这样不仅能够使得 hash 值能够分散能够均匀减小 hash 冲突的概率，另外只用到了异或运算，在性能开销上也能兼顾，做到平衡的 trade-off。 2、初始化 table 紧接着到第 2 步，会判断当前 table 数组是否初始化了，没有的话就调用 initTable 进行初始化，该方法在上面已经讲过了。 3、能否直接将新值插入到 table 数组中 从上面的结构示意图就可以看出存在这样一种情况，如果插入值待插入的位置刚好所在的 table 数组为 null 的话就可以直接将值插入即可。那么怎样根据 hash 确定在 table 中待插入的索引 i 呢？很显然可以通过 hash 值与数组的长度取模操作，从而确定新值插入到数组的哪个位置。而之前我们提过 ConcurrentHashMap 的大小总是 2 的幂次方，(n - 1) & hash 运算等价于对长度 n 取模，也就是 hash% n，但是位运算比取模运算的效率要高很多，Doug lea 大师在设计并发容器的时候也是将性能优化到了极致，令人钦佩。 确定好数组的索引 i 后，就可以可以 tabAt () 方法（该方法在上面已经说明了，有疑问可以回过头去看看）获取该位置上的元素，如果当前 Node f 为 null 的话，就可以直接用 casTabAt 方法将新值插入即可。 4、当前是否正在扩容 如果当前节点不为 null，且该节点为特殊节点（forwardingNode）的话，就说明当前 concurrentHashMap 正在进行扩容操作，关于扩容操作，下面会作为一个具体的方法进行讲解。那么怎样确定当前的这个 Node 是不是特殊的节点了？是通过判断该节点的 hash 值是不是等于 - 1（MOVED）, 代码为 (fh = f.hash) == MOVED，对 MOVED 的解释在源码上也写的很清楚了： static final int MOVED = -1; // hash for forwarding nodes 5、当 table [i] 为链表的头结点，在链表中插入新值 在 table [i] 不为 null 并且不为 forwardingNode 时，并且当前 Node f 的 hash 值大于 0（fh >= 0）的话说明当前节点 f 为当前桶的所有的节点组成的链表的头结点。那么接下来，要想向 ConcurrentHashMap 插入新值的话就是向这个链表插入新值。通过 synchronized (f) 的方式进行加锁以实现线程安全性。往链表中插入节点的部分代码为： if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; // 找到hash值相同的key,覆盖旧值即可 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; } } } 两种情况：1. 在链表中如果找到了与待插入的键值对的 key 相同的节点，就直接覆盖即可；2. 如果直到找到了链表的末尾都没有找到的话，就直接将待插入的键值对追加到链表的末尾即可 6、当 table [i] 为红黑树的根节点，在红黑树中插入新值 按照之前的数组 + 链表的设计方案，这里存在一个问题，即使负载因子和 Hash 算法设计的再合理，也免不了会出现拉链过长的情况，一旦出现拉链过长，甚至在极端情况下，查找一个节点会出现时间复杂度为 O (n) 的情况，则会严重影响 ConcurrentHashMap 的性能，于是，在 JDK1.8 版本中，对数据结构做了进一步的优化，引入了红黑树。而当链表长度太长（默认超过 8）时，链表就转换为红黑树，利用红黑树快速增删改查的特点提高 ConcurrentHashMap 的性能，其中会用到红黑树的插入、删除、查找等算法。当 table [i] 为红黑树的树节点时的操作为： 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 中通过 f instanceof TreeBin 判断当前 table [i] 是否是树节点，这下也正好验证了我们在最上面介绍时说的 TreeBin 会对 TreeNode 做进一步封装，对红黑树进行操作的时候针对的是 TreeBin 而不是 TreeNode。这段代码很简单，调用 putTreeVal 方法完成向红黑树插入新节点，同样的逻辑，如果在红黑树中存在于待插入键值对的 Key 相同（hash 值相等并且 equals 方法判断为 true）的节点的话，就覆盖旧值，否则就向红黑树追加新节点。 7、根据当前节点个数进行调整 当完成数据新节点插入之后，会进一步对当前链表大小进行调整，这部分代码为： if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } 很容易理解，如果当前链表节点个数大于等于 8（TREEIFY_THRESHOLD）的时候，就会调用 treeifyBin 方法将 tabel [i]（第 i 个散列桶）拉链转换成红黑树。 至此，关于 Put 方法的逻辑就基本说的差不多了，现在来做一些总结： 整体流程： 首先对于每一个放入的值，首先利用 spread 方法对 key 的 hashcode 进行一次 hash 计算，由此来确定这个值在 table 中的位置； 如果当前 table 数组还未初始化，先将 table 数组进行初始化操作； 如果这个位置是 null 的，那么使用 CAS 操作直接放入； 如果这个位置存在结点，说明发生了 hash 碰撞，首先判断这个节点的类型。如果该节点 fh==MOVED (代表 forwardingNode, 数组正在进行扩容) 的话，说明正在进行扩容； 如果是链表节点（fh>0）, 则得到的结点就是 hash 值相同的节点组成的链表的头节点。需要依次向后遍历确定这个新加入的值所在位置。如果遇到 key 相同的节点，则只需要覆盖该结点的 value 值即可。否则依次向后遍历，直到链表尾插入这个结点； 如果这个节点的类型是 TreeBin 的话，直接调用红黑树的插入方法进行插入新的节点； 插入完节点之后再次检查链表长度，如果长度大于 8，就把这个链表转换成红黑树； 对当前容量大小进行检查，如果超过了临界值（实际大小 * 加载因子）就需要扩容。 3.4、get 方法 看完了 put 方法再来看 get 方法就很容易了，用逆向思维去看就好，这样存的话我反过来这么取就好了。get 方法源码为： public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // 1. 重hash int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { // 2. table[i]桶节点的key与查找的key相同，则直接返回 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } // 3. 当前节点hash小于0说明为树节点，在红黑树中查找即可 else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) { //4. 从链表中查找，查找到则返回该节点的value，否则就返回null即可 if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; } 代码的逻辑请看注释，首先先看当前的 hash 桶数组节点即 table [i] 是否为查找的节点，若是则直接返回；若不是，则继续再看当前是不是树节点？通过看节点的 hash 值是否为小于 0，如果小于 0 则为树节点。如果是树节点在红黑树中查找节点；如果不是树节点，那就只剩下为链表的形式的一种可能性了，就向后遍历查找节点，若查找到则返回节点的 value 即可，若没有找到就返回 null。 3.5、transfer 方法 当 ConcurrentHashMap 容量不足的时候，需要对 table 进行扩容。这个方法的基本思想跟 HashMap 是很像的，但是由于它是支持并发扩容的，所以要复杂的多。原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足 concurrent 的要求，而是希望利用并发处理去减少扩容带来的时间影响。transfer 方法源码为： private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range //1. 新建Node数组，容量为之前的两倍 if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } int nextn = nextTab.length; //2. 新建forwardingNode引用，在之后会用到 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; // 3. 确定遍历中的索引i while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound; i = nextIndex - 1; advance = false; } } //4.将原数组中的元素复制到新数组中去 //4.5 for循环退出，扩容结束修改sizeCtl属性 if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) { nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1); return; } if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; finishing = advance = true; i = n; // recheck before commit } } //4.1 当前数组中第i个元素为null，用CAS设置成特殊节点forwardingNode(可以理解成占位符) else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); //4.2 如果遍历到ForwardingNode节点 说明这个点已经被处理过了 直接跳过 这里是控制并发扩容的核心 else if ((fh = f.hash) == MOVED) advance = true; // already processed else { synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; if (fh >= 0) { //4.3 处理当前节点为链表的头结点的情况，构造两个链表，一个是原链表 另一个是原链表的反序排列 int runBit = fh & n; Node<K,V> lastRun = f; for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) { ln = lastRun; hn = null; } else { hn = lastRun; ln = null; } for (Node<K,V> p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); } //在nextTable的i位置上插入一个链表 setTabAt(nextTab, i, ln); //在nextTable的i+n的位置上插入另一个链表 setTabAt(nextTab, i + n, hn); //在table的i位置上插入forwardNode节点 表示已经处理过该节点 setTabAt(tab, i, fwd); //设置advance为true 返回到上面的while循环中 就可以执行i--操作 advance = true; } //4.4 处理当前节点是TreeBin时的情况，操作和上面的类似 else if (f instanceof TreeBin) { TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> lo = null, loTail = null; TreeNode<K,V> hi = null, hiTail = null; int lc = 0, hc = 0; for (Node<K,V> e = t.first; e != null; e = e.next) { int h = e.hash; TreeNode<K,V> p = new TreeNode<K,V> (h, e.key, e.val, null, null); if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : (hc != 0) ? new TreeBin<K,V>(lo) : t; hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : (lc != 0) ? new TreeBin<K,V>(hi) : t; setTabAt(nextTab, i, ln); setTabAt(nextTab, i + n, hn); setTabAt(tab, i, fwd); advance = true; } } } } } } 代码逻辑请看注释，整个扩容操作分为两个部分： 第一部分是构建一个 nextTable, 它的容量是原来的两倍，这个操作是单线程完成的。新建 table 数组的代码为:Node<K,V>[] nt = (Node<K,V>[]) new Node<?,?>[n << 1], 在原容量大小的基础上右移一位。 第二个部分就是将原来 table 中的元素复制到 nextTable 中，主要是遍历复制的过程。 根据运算得到当前遍历的数组的位置 i，然后利用 tabAt 方法获得 i 位置的元素再进行判断： 1、如果这个位置为空，就在原 table 中的 i 位置放入 forwardNode 节点，这个也是触发并发扩容的关键点； 2、如果这个位置是 Node 节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把他们分别放在 nextTable 的 i 和 i+n 的位置上 3、如果这个位置是 TreeBin 节点（fh<0），也做一个反序处理，并且判断是否需要 untreefi，把处理的结果分别放在 nextTable 的 i 和 i+n 的位置上 4、遍历过所有的节点以后就完成了复制工作，这时让 nextTable 作为新的 table，并且更新 sizeCtl 为新容量的 0.75 倍 ，完成扩容。设置为新容量的 0.75 倍代码为 sizeCtl = (n << 1) - (n >>> 1)，仔细体会下是不是很巧妙，n<<1 相当于 n 右移一位表示 n 的两倍即 2n,n>>>1 左右一位相当于 n 除以 2 即 0.5n, 然后两者相减为 2n-0.5n=1.5n, 是不是刚好等于新容量的 0.75 倍即 2n*0.75=1.5n。最后用一个示意图来进行总结（图片摘自网络）： 3.6、与 size 相关的一些方法 对于 ConcurrentHashMap 来说，这个 table 里到底装了多少东西其实是个不确定的数量，因为不可能在调用 size () 方法的时候像 GC 的 “stop the world” 一样让其他线程都停下来让你去统计，因此只能说这个数量是个估计值。对于这个估计值，ConcurrentHashMap 也是大费周章才计算出来的。 为了统计元素个数，ConcurrentHashMap 定义了一些变量和一个内部类。 /** * A padded cell for distributing counts. Adapted from LongAdder * and Striped64. See their internal docs for explanation. */ @sun.misc.Contended static final class CounterCell { volatile long value; CounterCell(long x) { value = x; } } /******************************************/ /** * 实际上保存的是hashmap中的元素个数 利用CAS锁进行更新 但它并不用返回当前hashmap的元素个数 */ private transient volatile long baseCount; /** * Spinlock (locked via CAS) used when resizing and/or creating CounterCells. */ private transient volatile int cellsBusy; /** * Table of counter cells. When non-null, size is a power of 2. */ private transient volatile CounterCell[] counterCells; mappingCount 与 size 方法 mappingCount 与 size 方法的类似 从给出的注释来看，应该使用 mappingCount 代替 size 方法 两个方法都没有直接返回 basecount 而是统计一次这个值，而这个值其实也是一个大概的数值，因此可能在统计的时候有其他线程正在执行插入或删除操作。 public int size() { long n = sumCount(); return ((n < 0L) ? 0 : (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n); } /** * Returns the number of mappings. This method should be used * instead of {@link #size} because a ConcurrentHashMap may * contain more mappings than can be represented as an int. The * value returned is an estimate; the actual count may differ if * there are concurrent insertions or removals. * * @return the number of mappings * @since 1.8 */ public long mappingCount() { long n = sumCount(); return (n < 0L) ? 0L : n; // ignore transient negative values } final long sumCount() { CounterCell[] as = counterCells; CounterCell a; long sum = baseCount; if (as != null) { for (int i = 0; i < as.length; ++i) { if ((a = as[i]) != null) sum += a.value;//所有counter的值求和 } } return sum; } addCount方法 在put方法结尾处调用了addCount方法，把当前ConcurrentHashMap的元素个数+1这个方法一共做了两件事,更新baseCount的值，检测是否进行扩容。 private final void addCount(long x, int check) { CounterCell[] as; long b, s; //利用CAS方法更新baseCount的值 if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true; if (as == null || (m = as.length - 1) < 0 || (a = as[ThreadLocalRandom.getProbe() & m]) == null || !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { fullAddCount(x, uncontended); return; } if (check <= 1) return; s = sumCount(); } //如果check值大于等于0 则需要检验是否需要进行扩容操作 if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n); // if (sc < 0) { if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; //如果已经有其他线程在执行扩容操作 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } //当前线程是唯一的或是第一个发起扩容的线程 此时nextTable=null else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount(); } } } 四、总结 JDK6,7 中的 ConcurrentHashmap 主要使用 Segment 来实现减小锁粒度，分割成若干个 Segment，在 put 的时候需要锁住 Segment，get 时候不加锁，使用 volatile 来保证可见性，当要统计全局时（比如 size），首先会尝试多次计算 modcount 来确定，这几次尝试中，是否有其他线程进行了修改操作，如果没有，则直接返回 size。如果有，则需要依次锁住所有的 Segment 来计算。 1.8 之前 put 定位节点时要先定位到具体的 segment，然后再在 segment 中定位到具体的桶。而在 1.8 的时候摒弃了 segment 臃肿的设计，直接针对的是 Node [] tale 数组中的每一个桶，进一步减小了锁粒度。并且防止拉链过长导致性能下降，当链表长度大于 8 的时候采用红黑树的设计。 主要设计上的变化有以下几点: 1、不采用 segment 而采用 node，锁住 node 来实现减小锁粒度。 2、设计了 MOVED 状态 当 resize 的中过程中 线程 2 还在 put 数据，线程 2 会帮助 resize。 3、使用 3 个 CAS 操作来确保 node 的一些操作的原子性，这种方式代替了锁。 4、sizeCtl 的不同值来代表不同含义，起到了控制的作用。 5、采用 synchronized 而不是 ReentrantLock。
一、简介 在之前介绍 AQS 的底层实现，已经在介绍 java 中的 Lock 时，比如 ReentrantLock,ReentReadWriteLocks, 已经在介绍线程间等待 / 通知机制使用的 Condition 时都会调用 LockSupport.park () 方法和 LockSupport.unpark () 方法。LockSupport 位于 java.util.concurrent.locks 包下，该类的方法并不是很多。LockSupprot 是线程的阻塞原语，用来阻塞线程和唤醒线程。每个使用 LockSupport 的线程都会与一个许可关联，如果该许可可用，并且可在线程中使用，则调用 park () 将会立即返回，否则可能阻塞。如果许可尚不可用，则可以调用 unpark 使其可用。但是注意许可不可重入，也就是说只能调用一次 park () 方法，否则会一直阻塞。 二、方法介绍 2.1、阻塞线程 1、void park ()：阻塞当前线程，如果调用 unpark 方法或者当前线程被中断，从能从 park () 方法中返回 2、void park (Object blocker)：功能同方法 1，入参增加一个 Object 对象，用来记录导致线程阻塞的阻塞对象，方便进行问题排查； 3、void parkNanos (long nanos)：阻塞当前线程，最长不超过 nanos 纳秒，增加了超时返回的特性； 4、void parkNanos (Object blocker, long nanos)：功能同方法 3，入参增加一个 Object 对象，用来记录导致线程阻塞的阻塞对象，方便进行问题排查； 5、void parkUntil (long deadline)：阻塞当前线程，知道 deadline； 6、void parkUntil (Object blocker, long deadline)：功能同方法 5，入参增加一个 Object 对象，用来记录导致线程阻塞的阻塞对象，方便进行问题排查； 2.2、唤醒线程 void unpark (Thread thread): 唤醒处于阻塞状态的指定线程。 实际上 LockSupport 阻塞和唤醒线程的功能是依赖于 sun.misc.Unsafe，这是一个很底层的类，有兴趣的可以去查阅资料，比如 park () 方法的功能实现则是靠 unsafe.park () 方法。另外在阻塞线程这一系列方法中还有一个很有意思的现象就是，每个方法都会新增一个带有 Object 的阻塞对象的重载方法。那么增加了一个 Object 对象的入参会有什么不同的地方了？示例代码很简单，直接看 dump 线程的信息。 2.3、调用 park () 方法 dump 线程 "main" #1 prio=5 os_prio=0 tid=0x02cdcc00 nid=0x2b48 waiting on condition [0x00d6f000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:304) at learn.LockSupportDemo.main(LockSupportDemo.java:7) 2.4、调用 park (Object blocker) 方法 dump 线程 "main" #1 prio=5 os_prio=0 tid=0x0069cc00 nid=0x6c0 waiting on condition [0x00dcf000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for <0x048c2d18> (a java.lang.String) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at learn.LockSupportDemo.main(LockSupportDemo.java:7) 通过分别调用这两个方法然后 dump 线程信息可以看出，带 Object 的 park 方法相较于无参的 park 方法会增加 parking to wait for <0x048c2d18> (a java.lang.String）的信息，这种信息就类似于记录 “案发现场”，有助于工程人员能够迅速发现问题解决问题。有个有意思的事情是，我们都知道如果使用 synchronzed 阻塞了线程 dump 线程时都会有阻塞对象的描述，在 java 5 推出 LockSupport 时遗漏了这一点，在 java 6 时进行了补充。还有一点需要需要的是：synchronzed 致使线程阻塞，线程会进入到 BLOCKED 状态，而调用 LockSupprt 方法阻塞线程会致使线程进入到 WAITING 状态。 三、一个例子 public class LockSupportDemo { public static void main(String[] args) { Thread thread = new Thread(() -> { LockSupport.park(); System.out.println(Thread.currentThread().getName() + "被唤醒"); }); thread.start(); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } LockSupport.unpark(thread); } } thread 线程调用 LockSupport.park () 致使 thread 阻塞，当 mian 线程睡眠 3 秒结束后通过 LockSupport.unpark (thread) 方法唤醒 thread 线程，thread 线程被唤醒执行后续操作。另外，还有一点值得关注的是，LockSupport.unpark (thread) 可以指定线程对象唤醒指定的线程。
一、Condition 简介 任何一个 java 对象都天然继承于 Object 类，在线程间实现通信的往往会应用到 Object 的几个方法，比如 wait (),wait (long timeout),wait (long timeout, int nanos) 与 notify (),notifyAll () 几个方法实现等待 / 通知机制，同样的， 在 java Lock 体系下依然会有同样的方法实现等待 / 通知机制。从整体上来看 Object 的 wait 和 notify/notify 是与对象监视器配合完成线程间的等待 / 通知机制，而 Condition 与 Lock 配合完成等待通知机制，前者是 java 底层级别的，后者是语言级别的，具有更高的可控制性和扩展性。两者除了在使用方式上不同外，在功能特性上还是有很多的不同： 1、Condition 能够支持不响应中断，而通过使用 Object 方式不支持； 2、Condition 能够支持多个等待队列（new 多个 Condition 对象），而 Object 方式只能支持一个； 3、Condition 能够支持超时时间的设置，而 Object 不支持 参照 Object 的 wait 和 notify/notifyAll 方法，Condition 也提供了同样的方法： 针对 Object 的 wait 方法 void await() throws InterruptedException 当前线程进入等待状态，如果其他线程调用 condition 的 signal 或者 signalAll 方法并且当前线程获取 Lock 从 await 方法返回，如果在等待状态中被中断会抛出被中断异常； long awaitNanos(long nanosTimeout) 当前线程进入等待状态直到被通知，中断或者超时； boolean await(long time, TimeUnit unit)throws InterruptedException 同第二种，支持自定义时间单位: boolean awaitUntil(Date deadline) throws InterruptedException 当前线程进入等待状态直到被通知，中断或者到了某个时间。 针对 Object 的 notify/notifyAll 方法 1、void signal ()：唤醒一个等待在 condition 上的线程，将该线程从等待队列中转移到同步队列中，如果在同步队列中能够竞争到 Lock 则可以从等待方法中返回。 2、void signalAll ()：与 1 的区别在于能够唤醒所有等待在 condition 上的线程 二、Condition 实现原理分析 2.1、等待队列 创建一个 condition 对象是通过 lock.newCondition (), 而这个方法实际上是会 new 出一个 ConditionObject 对象，该类是 AQS 的一个内部类。condition 是要和 lock 配合使用的也就是 condition 和 Lock 是绑定在一起的，而 lock 的实现原理又依赖于 AQS，自然而然 ConditionObject 作为 AQS 的一个内部类无可厚非。在锁机制的实现上，AQS 内部维护了一个同步队列，如果是独占式锁的话，所有获取锁失败的线程的尾插入到同步队列，同样的，condition 内部也是使用同样的方式，内部维护了一个 等待队列，所有调用 condition.await 方法的线程会加入到等待队列中，并且线程状态转换为等待状态。另外注意到 ConditionObject 中有两个成员变量： /** First node of condition queue. */ private transient Node firstWaiter; /** Last node of condition queue. */ private transient Node lastWaiter; ConditionObject 通过持有等待队列的头尾指针来管理等待队列。主要注意的是 Node 类复用了在 AQS 中的 Node 类。Node 类有这样一个属性： //后继节点 Node nextWaiter; 进一步说明，等待队列是一个单向队列，而在之前说 AQS 时知道同步队列是一个双向队列。接下来用一个 demo，通过 debug 进去看是不是符合猜想： public static void main(String[] args) { for (int i = 0; i < 10; i++) { Thread thread = new Thread(() -> { lock.lock(); try { condition.await(); } catch (InterruptedException e) { e.printStackTrace(); }finally { lock.unlock(); } }); thread.start(); } } 2.2、await 实现原理 当调用 condition.await () 方法后会使得当前获取 lock 的线程进入到等待队列，如果该线程能够从 await () 方法返回的话一定是该线程获取了与 condition 相关联的 lock。 public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 1. 将当前线程包装成Node，尾插入到等待队列中 Node node = addConditionWaiter(); // 2. 释放当前线程所占用的lock，在释放的过程中会唤醒同步队列中的下一个节点 int savedState = fullyRelease(node); int interruptMode = 0; while (!isOnSyncQueue(node)) { // 3. 当前线程进入到等待状态 LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } // 4. 自旋等待获取到同步状态（即获取到lock） if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) // clean up if cancelled unlinkCancelledWaiters(); // 5. 处理被中断的情况 if (interruptMode != 0) reportInterruptAfterWait(interruptMode); } 当前线程调用 condition.await () 方法后，会使得当前线程释放 lock 然后加入到等待队列中，直至被 signal/signalAll 后会使得当前线程从等待队列中移至到同步队列中去，直到获得了 lock 后才会从 await 方法返回，或者在等待时被中断会做中断处理。几个问题：1. 是怎样将当前线程添加到等待队列中去的？2. 释放锁的过程？3. 怎样才能从 await 方法退出？而这段代码的逻辑就是告诉我们这三个问题的答案。具体请看注释，在第 1 步中调用 addConditionWaiter 将当前线程添加到等待队列中，该方法源码为： private Node addConditionWaiter() { Node t = lastWaiter; // If lastWaiter is cancelled, clean out. if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } // 将当前线程包装成Node Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) firstWaiter = node; else // 尾插入 t.nextWaiter = node; //更新lastWaiter lastWaiter = node; return node; } 这段代码就很容易理解了，调用 AQS 的模板方法 release 方法释放 AQS 的同步状态并且唤醒在同步队列中头结点的后继节点引用的线程，如果释放成功则正常返回，若失败的话就抛出异常。到目前为止，这两段代码已经解决了前面的两个问题的答案了，还剩下第三个问题，怎样从 await 方法退出？现在回过头再来看 await 方法有这样一段逻辑： final int fullyRelease(Node node) { boolean failed = true; try { int savedState = getState(); if (release(savedState)) { // 成功释放同步状态 failed = false; return savedState; } else { // 不成功释放同步状态抛出异常 throw new IllegalMonitorStateException(); } } finally { if (failed) node.waitStatus = Node.CANCELLED; } } 这段代码就很容易理解了，将当前节点包装成 Node，如果等待队列的 firstWaiter 为 null 的话（等待队列为空队列），则将 firstWaiter 指向当前的 Node, 否则，更新 lastWaiter (尾节点) 即可。就是通过尾插入的方式将当前线程封装的 Node 插入到等待队列中即可，同时可以看出等待队列是一个不带头结点的链式队列，之前我们学习 AQS 时知道同步队列是一个带头结点的链式队列，这是两者的一个区别。将当前节点插入到等待对列之后，会使当前线程释放 lock，由 fullyRelease 方法实现，fullyRelease 源码为： while (!isOnSyncQueue(node)) { // 3. 当前线程进入到等待状态 LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } 很显然，当线程第一次调用 condition.await () 方法时，会进入到这个 while () 循环中，然后通过 LockSupport.park (this) 方法使得当前线程进入等待状态，那么要想退出这个 await 方法第一个前提条件自然而然的是要先退出这个 while 循环，出口就只剩下两个地方：1. 逻辑走到 break 退出 while 循环；2. while 循环中的逻辑判断为 false。再看代码出现第 1 种情况的条件是当前等待的线程被中断后代码会走到 break 退出，第二种情况是当前节点被移动到了同步队列中（即另外线程调用的 condition 的 signal 或者 signalAll 方法），while 中逻辑判断为 false 后结束 while 循环。总结下，就是当前线程被中断或者调用 condition.signal/condition.signalAll 方法当前节点移动到了同步队列后 ，这是当前线程退出 await 方法的前提条件。当退出 while 循环后就会调用 acquireQueued (node, savedState)，这个方法在介绍 AQS 的底层实现时说过了，若感兴趣的话可以去看这篇文章，该方法的作用是在自旋过程中线程不断尝试获取同步状态，直至成功（线程获取到 lock）。这样也说明了退出 await 方法必须是已经获得了 condition 引用（关联）的 lock。到目前为止，开头的三个问题我们通过阅读源码的方式已经完全找到了答案，也对 await 方法的理解加深。await 方法示意图如下图： 如图，调用 condition.await 方法的线程必须是已经获得了 lock，也就是当前线程是同步队列中的头结点。调用该方法后会使得当前线程所封装的 Node 尾插入到等待队列中。 condition 还额外支持了超时机制，使用者可调用方法 awaitNanos,awaitUtil。这两个方法的实现原理，基本上与 AQS 中的 tryAcquire 方法如出一辙，关于 tryAcquire 可以仔细阅读本文的第 3.4 部分。 不响应中断的支持： 要想不响应中断可以调用 condition.awaitUninterruptibly () 方法，该方法的源码为： public final void awaitUninterruptibly() { Node node = addConditionWaiter(); int savedState = fullyRelease(node); boolean interrupted = false; while (!isOnSyncQueue(node)) { LockSupport.park(this); if (Thread.interrupted()) interrupted = true; } if (acquireQueued(node, savedState) || interrupted) selfInterrupt(); } 这段方法与上面的 await 方法基本一致，只不过减少了对中断的处理，并省略了 reportInterruptAfterWait 方法抛被中断的异常。 2.3、signal/signalAll 实现原理 调用 condition 的 signal 或者 signalAll 方法可以将等待队列中等待时间最长的节点移动到同步队列中，使得该节点能够有机会获得 lock。按照等待队列是先进先出（FIFO）的，所以等待队列的头节点必然会是等待时间最长的节点，也就是每次调用 condition 的 signal 方法是将头节点移动到同步队列中。我们来通过看源码的方式来看这样的猜想是不是对的，signal 方法源码为： public final void signal() { //1. 先检测当前线程是否已经获取lock if (!isHeldExclusively()) throw new IllegalMonitorStateException(); //2. 获取等待队列中第一个节点，之后的操作都是针对这个节点 Node first = firstWaiter; if (first != null) doSignal(first); } signal 方法首先会检测当前线程是否已经获取 lock，如果没有获取 lock 会直接抛出异常，如果获取的话再得到等待队列的头指针引用的节点，之后的操作的 doSignal 方法也是基于该节点。下面我们来看看 doSignal 方法做了些什么事情，doSignal 方法源码为： private void doSignal(Node first) { do { if ( (firstWaiter = first.nextWaiter) == null) lastWaiter = null; //1. 将头结点从等待队列中移除 first.nextWaiter = null; //2. while中transferForSignal方法对头结点做真正的处理 } while (!transferForSignal(first) && (first = firstWaiter) != null); } 具体逻辑请看注释，真正对头节点做处理的逻辑在 transferForSignal 方法，该方法源码为： final boolean transferForSignal(Node node) { /* * If cannot change waitStatus, the node has been cancelled. */ //1. 更新状态为0 if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; /* * Splice onto queue and try to set waitStatus of predecessor to * indicate that thread is (probably) waiting. If cancelled or * attempt to set waitStatus fails, wake up to resync (in which * case the waitStatus can be transiently and harmlessly wrong). */ //2.将该节点移入到同步队列中去 Node p = enq(node); int ws = p.waitStatus; if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread); return true; } 关键逻辑请看注释，这段代码主要做了两件事情 1. 将头结点的状态更改为 CONDITION；2. 调用 enq 方法，将该节点尾插入到同步队列中，关于 enq 方法请看 AQS 的底层实现这篇文章。现在我们可以得出结论：调用 condition 的 signal 的前提条件是当前线程已经获取了 lock，该方法会使得等待队列中的头节点即等待时间最长的那个节点移入到同步队列，而移入到同步队列后才有机会使得等待线程被唤醒，即从 await 方法中的 LockSupport.park (this) 方法中返回，从而才有机会使得调用 await 方法的线程成功退出。signal 执行示意图如下图： sigllAll 与 sigal 方法的区别体现在 doSignalAll 方法上，前面我们已经知道 doSignal 方法只会对等待队列的头节点进行操作，而 doSignalAll 的源码为： private void doSignalAll(Node first) { lastWaiter = firstWaiter = null; do { Node next = first.nextWaiter; first.nextWaiter = null; transferForSignal(first); first = next; } while (first != null); } 该方法只不过时间等待队列中的每一个节点都移入到同步队列中，即 “通知” 当前调用 condition.await () 方法的每一个线程。 三、await 与 signal/signalAll 的结合思考 public class AwaitSignal { private static ReentrantLock lock = new ReentrantLock(); private static Condition condition = lock.newCondition(); private static volatile boolean flag = false; public static void main(String[] args) { Thread waiter = new Thread(new waiter()); waiter.start(); Thread signaler = new Thread(new signaler()); signaler.start(); } static class waiter implements Runnable { @Override public void run() { lock.lock(); try { while (!flag) { System.out.println(Thread.currentThread().getName() + "当前条件不满足等待"); try { condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println(Thread.currentThread().getName() + "接收到通知条件满足"); } finally { lock.unlock(); } } } static class signaler implements Runnable { @Override public void run() { lock.lock(); try { flag = true; condition.signalAll(); } finally { lock.unlock(); } } } } 输出结果为： Thread-0 当前条件不满足等待 Thread-0 接收到通知，条件满足 开启了两个线程 waiter 和 signaler，waiter 线程开始执行的时候由于条件不满足，执行 condition.await 方法使该线程进入等待状态同时释放锁，signaler 线程获取到锁之后更改条件，并通知所有的等待线程后释放锁。这时，waiter 线程获取到锁，并由于 signaler 线程更改了条件此时相对于 waiter 来说条件满足，继续执行。