桥下の民 - 并发编程
type: channel
link: https://wordplay.work/index.php/category/concurrent
items: 并发编程 Code
items: CountDownLatch
items: BlockingQueue 阻塞队列
items: ThreadLocal 内存泄漏
items: ThreadLocal 详解
items: 容器 CopyOnWriteArrayList
items: 容器 ConcurrentLinkedQueue
items: 容器 ConcurrentHashmap
items: LockSupport 工具
items: Condition 的 await & signal

并发编程 Code
type: item
date: 2024-03-02T02:46:00+00:00
description: 一、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…

CountDownLatch
type: item
date: 2023-03-12T05:34:00+00:00
description: 一、简介 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(); } } }

BlockingQueue 阻塞队列
type: item
date: 2022-05-10T06:11:00+00:00
description: 一、简介 在实际编程中，会经常使用到 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 是一个支持优先级的无界阻塞队列。默认情况下元素采用自然顺序进行排序，也可以通过自定义类实现 comp…

ThreadLocal 内存泄漏
type: item
date: 2022-05-10T03:35:00+00:00
description: 一、原因 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 …

ThreadLocal 详解
type: item
date: 2022-05-10T01:27:00+00:00
description: 一、简介 在多线程编程中通常解决线程安全的问题我们会利用 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) { @Supp…

容器 CopyOnWriteArrayList
type: item
date: 2022-05-09T03:53:00+00:00
description: 一、简介 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 操作等等…

容器 ConcurrentLinkedQueue
type: item
date: 2022-05-08T14:04:00+00:00
description: 一、简介 在单线程编程中会经常用到一些集合类，比如 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 线程两者并无 “交集”，也就是说两类线程间并不会相…

容器 ConcurrentHashmap
type: item
date: 2022-04-29T13:37:00+00:00
description: 一、简介 在使用 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…

LockSupport 工具
type: item
date: 2022-04-29T02:40:00+00:00
description: 一、简介 在之前介绍 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）的信息，这种信息就类似于记录 “案发现场”，有助于工程人员能够迅…

Condition 的 await & signal
type: item
date: 2022-04-27T14:22:00+00:00
description: 一、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(() -> { …