线程基础

文章目录

线程基础创建线程以及运行继承Thread类实现Runnable接口实现Callable接口线程的通知和等待wait()函数notify()和notifyAll()函数等待线程执行终止的join()函数使线程睡眠的sleep()函数使CPU让出执行权的yield()函数线程中断线程的上下文切换线程死锁守护线程和用户线程ThreadLocal实现原理key为弱引用重点多线程并发编程并发和并行多线程并发编程的作用设计共享变量的内存可见性问题CAS操作指令重排序伪共享锁乐观锁和悲观锁公平锁和非公平锁独占锁和共享锁可重入锁自旋锁高级部分ThreadLocalRandom类原子操作类AtomicLongLongAdderSynchronized专题Java并发包中锁原理LockSupport工具类抽象同步队列AQSCountDownLatch实现原理ReentrantLock互斥锁源码逐行分析一、提一嘴CAS二、简单阐述AQS核心内容三、查看lock方法（公平锁和非公平锁）四、查看acquire方法4.1 执行流程4.1.1 tryAcquire执行流程4.1.2 addWaiter方法4.1.3 acquireQueued方法流程4.2 tryAcquire4.2.1 非公平锁4.2.2 公平锁4.3 addWaiter4.4 acquireQueued五、查看unlock方法ThreadPoolExecutor源码分析JDK自带的线程池一、线程池的核心属性二、线程池的7个参数三、线程池的执行流程（execute方法）四、线程池如何添加工作线程（addWorker方法）五、工作线程执行任务方式（runWorker方法）六、工作线程获取任务方式&结束工作线程方式（getTask方法）一般线程池参数设置多少？自带的拒绝策略？都是工作线程，到时间，怎么知道这个被销毁，那个不销毁？并发集合ConcurrentHashMap一、存储结构二、存储操作2.1 put方法2.2 putVal方法-散列算法2.3 putVal方法-添加数据到数组&初始化数组2.4 putVal方法-添加数据到链表三、扩容操作3.1 treeifyBin方法触发扩容3.2 tryPreSize方法-针对putAll的初始化操作3.3 tryPreSize方法-计算扩容戳并且查看BUG3.4 tryPreSize方法-对sizeCtl的修改以及条件判断的BUG3.5 transfer方法-计算每个线程迁移的长度3.6 transfer方法-构建新数组并查看标识属性3.7 transfer方法-线程领取迁移任务3.8 transfer方法-迁移结束操作3.9 transfer方法-迁移数据（链表）3.10 helpTransfer方法-协助扩容四、红黑树操作4.1 什么是红黑树4.2 TreeifyBin方法-封装TreeNode和双向链表4.3 TreeBin有参构造-双向链表转为红黑树4.4 balanceInsertion方法-保证红黑树平衡以及特性4.5 putTreeVal方法-添加节点4.6 TreeBin的锁操作4.7 transfer迁移数据五、查询数据5.1 get方法-查询数据的入口5.2 ForwardingNode的find方法5.3 ReservationNode的find方法5.4 TreeBin的find方法5.6 TreeNode的findTreeNode方法六、**ConcurrentHashMap其他方法**6.1 compute方法6.2 compute方法源码分析6.3 computeIfPresent、computeIfAbsent、compute区别6.4 replace方法详解6.5 merge方法详解七、**ConcurrentHashMap计数器**一、addCount方法分析二、size方法方法分析

创建线程以及运行

结论:三种方式

继承Thread类实现Runnable接口实现Callable接口

继承Thread类

创建代码:

public class MyThread extends Thread {@Overridepublic void run() {System.out.println("继承Thread类创建线程");}}

运行代码:

public class Test {public static void main(String[] args) {//创建线程MyThread myThread = new MyThread();//启动一个线程myThread.start();}}

上面代码继承了Thread类,重写了run方法。在main函数中创建一个实例对象，并且调取这个实例对象的start方法，线程才正式启动。

其实调用start方法后，并没有立即执行而是处于就绪状态，等待获取CPU资源后才会开始运行，run方法执行完后，这个线程也会终止。

使用继承的好处：在run方法里获取获取当前线程直接使用this就可以，不需要调用Thread.currentThread()方法

缺点就是Java语言为单继承方式,不能其他类了;还有就是多个线程执行一样任务时需要多份任务代码

实现Runnable接口

创建代码:

public class RunableTask implements Runnable {@Overridepublic void run() {System.out.println("通过实现Runnable接口创建线程");}}

运行代码:

public class Test {public static void main(String[] args) {RunableTask task = new RunableTask();new Thread(task).start();new Thread(task).start();}}

实现Runnable接口,重写run方法,两个线程公用一个task代码,也可继承和实现其他类或接口,但是上面两种方式都有一个缺点没有返回值

实现Callable接口

创建代码:

public class CallTask implements Callable<String> {@Overridepublic String call() throws Exception {return "通过实现callable接口创建线程";}}

运行代码:

public class Test {public static void main(String[] args) {FutureTask<String> futureTask = new FutureTask<>(new CallTask());new Thread(futureTask).start();try {String s = futureTask.get();System.out.println(s);} catch (InterruptedException e) {throw new RuntimeException(e);} catch (ExecutionException e) {throw new RuntimeException(e);}}

上面的代码实现了Callable接口,重写call方法。在main函数中创建FutureTask对象(构建CallTask的实例),然后通过创建的Future作为任务创建一个线程启动它,最后通过future.get()等待任务执行完毕获取返回结果,前两中方法都没办法获取返回值,但是最后一直通过FutureTask对象获取返回值。

线程的通知和等待

object类是java中的父类,鉴于继承机制,java把所有类所需要的方法抽取到Object类(hashcode,equals,clone,finalize,wait,notify,notifyall,toString)

wait()函数

当一个线程调用了共享变量的wait()方法后,该线程会被阻塞起来,知道反生下面两件事之一线程才进入就绪状态

调用notify或者notifyall方法其他线程调用了该线程的interrupt()方法,该线程抛出溢出

使用姿势:

public class Demo1 {Object object = new Object();int num = 0;public void test1(){synchronized(object){while(num<0){try {object.wait();} catch (InterruptedException e) {throw new RuntimeException(e);}}}}}

原理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VT2EqFTq-1666103600147)(C:\Users\小刘同学\AppData\Roaming\Typora\typora-user-images\image-0807192455866.png)]

结合上面代码,线程一和线程二发现自己条件不满足,共享变量调用wait()方法,将线程一个线程二进入WaitSet队列,WaitSet里面的线程等待Owner中的线程调用notify或notifyAll方法,告诉WaitSet队列里的线程条件满足来恢复到就绪状态,参与CPU的争夺

notify()和notifyAll()函数

notify():当锁持有者Owner调用notify函数,告诉Waitset中的线程条件满足,随机挑选一个线程恢复到就绪队列中

notifyAll():将Waitset中的所有线程释放添加到就绪队列中

等待线程执行终止的join()函数

在项目场景中,就是需要等待某几件事完成后才能继续执行

比如加载资源,多线程下进行资源处理,等资源全部加载完后再进行汇总

Thread类中就有这么一个Join方法,前面wait/notify是Object中的方法

join函数是一个无参无返回值的函数

使用姿势:

public static void main(String[] args) throws InterruptedException {Thread threadOne = new Thread(new Runnable() {@Overridepublic void run() {System.out.println("这是子线程一");}});Thread threadTwo = new Thread(new Runnable() {@Overridepublic void run() {System.out.println("这是子线程二");}});//启动子线程threadOne.start();Thread.sleep(1000);threadTwo.start();//主线程等待子线程执行完毕threadOne.join();threadTwo.join();System.out.println("子线程执行完毕");}

在主线程中创建两个子线程,分别调用join方法,那么主线程首先会在调用 threadOne.join();阻塞,直到threadOne运行结束后主线程才运行,

这里演示Join的使用方法,在JUC包中的CountDownLatch是一个不错的选择

使线程睡眠的sleep()函数

在执行的一个线程中执行了sleep函数,使当前线程进入到了一个有时等待,比如在两秒后,线程就会让出指定时间的CPU执行权,也就是线程在这段时间不执行

面试重点

wait和sleep的区别

wait是Object中的一个方法,用于线程之间的通讯;sleep是Thread中的一个静态方法wait必须配合同步代码块进行使用,是因为在一个线程进入同步代码块时,会关联一个Monitor对象,如果条件不满足,关联对象调用wait方法进入waitSet队列,必须等待notify/notifyAll唤醒,不然意思就是条件一直不满足,一直在waitset中,这时会释放锁sleep让当前执行的线程进入一个有时等待状态,在这个状态时间里,让出了CPU的执行权,如果不存在其他线程进行CPU的竞争,则不会释放锁,如果有,则进入就绪队列重写进行CPU的竞争两者都可以对线程暂停运行

使CPU让出执行权的yield()函数

当一个线程执行yield函数,就会让出CPU的执行权,进入就绪队列,表示当前线程进去CPU下一轮的调度。

正常的情况下是CPU给当前线程分配时间片，当时间片执行完后该线程进入下一轮的线程的调度。当线程执行了静态方法yield()函数后，该线程就进入就绪队列，让出CPU的执行权。

线程中断

void interrupt()方法：中断线程，当线程A正在执行时，线程B调用了线程A的interrupt方法来设置线程A的标志位true，实际上线程A并没有中断，只是仅仅设置了中断标志，还会继续向下执行。如果线程A调用了wait函数，join，sleep函数而被阻塞挂起，那么调用interrupt会抛出异常

boolean isInterrupted()方法：检测当前线程是否被中断

boolean interrupted()方法：检测线程是否被中断，这个与isInterrupted不同的是该方法为static静态方法，可以通过Thread直接调用，如果发现当前线程被中断，则会清除中断标志。

线程的上下文切换

这要分为两种情况

切换的线程属于同一个进程

因为虚拟内存是共享的，只切换线程私有的部分，所以比如程序计数器，栈信息的交换。比进程切换开销小了好多

切换的线程属于不同进程

由于属于不同进程，所以说线程的切换属于进程切换，这个切换开销就比较大

首先要保持当前CPU的信息（就是当前执行的线程），比如程序计数器和栈信息更新PCB进程控制块信息将PCB放入到对应的队列当中，比如就绪队列和阻塞队列中CPU调取下一个进程，更新当前的CPU信息更新内存中的数据恢复CPU的上下文处理

线程死锁

在多线程下，多个线程竞争一个临界资源（共享数据）时，该资源假设只能被一个线程占有，当一个线程持有了这个资源并去请求另外一个资源时，其他线程也是要获取这个资源，这些线程在获得的资源情况下，只能自己去释放资源，不能强迫被释放，在这些条件下，会发生一个死锁

互斥条件请求和保持不可剥夺循环等待

检测发生死锁：jps 列出当前运行的java程序，并且列出进程好

jstack命令查看进程信息

守护线程和用户线程

Java线程分为两种线程，守护线程和用户线程，在JVM启动时，会调用一个main函数，main函数所在的线程就是一个用户线程，同时在JVM内部也启动了好多守护线程，比如垃圾回收线程。

当最后一个用户线程执行结束后,程序运行结束,不管当前是否存在守护线程,守护线程不影响程序的运行结束

设置为守护线程setDaemon

ThreadLocal

简介:为线程设置一份本地变量,其他变量不能进行访问,当多线程操作这个变量时,实际上是访问本地内存里面里面的变量,从而避免了线程安全问题;

如果创建了一个ThreadLocal变量,那么访问这个变量的线程都有一个这个变量的本地副本.当多个线程操作这个变量时,操作的也是自己本地内存的变量.

使用规范:

public class Test {static ThreadLocal<Integer> nums = new ThreadLocal<>();static void print(String str){//打印当前线程的本地变量值System.out.println(str+":"+nums.get());}public static void main(String[] args) {new Thread(()->{nums.set(10);print("线程一");nums.set(nums.get()-1);System.out.println("线程一之后的变量值"+nums.get());}).start();new Thread(()->{nums.set(15);print("线程二");System.out.println("线程二之后的变量值"+nums.get());}).start();}}

实现原理

类图分析

在Thread类中有两个ThreadLocalMap的变量,是threadLocals和inheritableThreadLocals。而ThreadLocalMap是HashMap的定制版。在默认情况下，每个线程的这两个变量都为null，只有当前使用的线程第一次调用ThreadLocal的set方法和get方法时才会创建。每个线程的本地变量不是存放在ThreadLocal实例中，而是存放在调用线程的threadLocals变量里。ThreadLocal实际上就是一个工具壳，通过set方法把value值放入到threadlocals里面存放起来。当不使用本地变量时，调用remove方法，从而使threadLocals中的变量删除

为什么threadLocals设计为map结构

这个设计可以使线程存储多个本地变量

对set，get，remove方法进行分析

1.void set（T value）

public void set(T value) {//获取当前线程Thread t = Thread.currentThread();//获取当前线程的threadLocals变量ThreadLocalMap map = getMap(t);//懒加载机制 如果这个变量为null 就进行设置否则直接改变这值，当前threadLocal作为keyif (map != null)map.set(this, value);else//以当前threadLocal作为key创建mapcreateMap(t, value);}//getMap方法 获取当前线程的threadLocalsThreadLocalMap getMap(Thread t) {return t.threadLocals;}//创建Map对象void createMap(Thread t, T firstValue) {t.threadLocals = new ThreadLocalMap(this, firstValue);}

首先获取当前的运行线程（Thread.currentThread()）;获取当前线程中的threadLocals变量（ThreadLocalMap类型的变量）判断这个变量是否为空 如果这个变量为空，通过ThreadLocalMap的构造方法进行set值，当前threadLocal为Key设置value如果不为空，通过ThreadLocalMap的set属性进行设置值

2.T get();

public T get() {//获取当前线程Thread t = Thread.currentThread();//获取当前线程的threadLocals变量ThreadLocalMap map = getMap(t);//如果不为空，就返回本地变量if (map != null) {ThreadLocalMap.Entry e = map.getEntry(this);if (e != null) {@SuppressWarnings("unchecked")T result = (T)e.value;return result;}}//如果为空，返回初始化当前的本地变量return setInitialValue();}private T setInitialValue() {//初始化为nullT value = initialValue();//获取当前线程Thread t = Thread.currentThread();ThreadLocalMap map = getMap(t);if (map != null)map.set(this, value);elsecreateMap(t, value);return value;}protected T initialValue() {return null;}

获取当前的线程获取当前线程的threadLocals变量判断当前map是否为null 如果为null，就对map进行初始化，初始化的值为null反则返回value

3.void remove()

public void remove() {ThreadLocalMap m = getMap(Thread.currentThread());if (m != null)m.remove(this);}

由于ThreadLocal不支持继承，所以子线程获取不到父线程的本地变量，因此推出了inheritableThreadLocals

key为弱引用

static class Entry extends WeakReference<ThreadLocal<?>> {/** The value associated with this ThreadLocal. */Object value;Entry(ThreadLocal<?> k, Object v) {super(k);value = v;}}

Entry节点设置Key为弱引用，避免了ThreadLocal对象无法被回收

假设设置为强引用，当前线程运行结束，对应的栈帧也随之销毁，但是现在Entry节点的key强引用为ThreadLocal，导致这个ThreadLocal对象无法被回收，造成内存泄漏

重点

ThreadLocalMap的Entry中的key使用的是弱引用，这是进一步防止内存泄漏，如果是强引用，所以的线程都没有对ThreadLocal引用，这个ThreadLocal也不会被回收，因为内部ThreadLocalMap也对ThreadLocal对象引用，这时候ThreadLocal是没有用的，也不会被垃圾回收。而如果是若引用会被回收，但是回收后存储的entry节点中的key值为null。虽然ThreadLocalMap提供了set，get，remove可以在一些时机对这些Entry进行清除，但是还是不及时的，比如在线程池中使用造成的内存泄漏。所以在使用完毕后调用remove方法才是解决内存泄漏的基本问题。

多线程并发编程

并发和并行

并发：首先在单核环境下，也就是当前只允许一个线程运行，多个线程会隔一段时间交换执行

并行：在多核环境下，多个线程同时执行

多线程并发编程的作用

异步性

同步:需要等待结果返回,才能继续执行

异步:不需要等待结果返回,就可以继续运行

设计

多线程可以让方法执行变为异步(不等待别的阻塞线程运行结束),比如读取磁盘文件,假设读取操作花了10秒,没有线程调度机制,这5秒调用者什么都做不了,其他代码都要等待

在项目中,上传大型文件比较费时,这时开启一个新线程专门用于处理文件操作,避免阻塞主线程的运行Tomcat的异步Servlet也是类似的,用户线程处理耗时较长的操作,避免阻塞Tomcat的工作线程

提高了效率

利用多核cpu的优势,提高运行效率

比如:任务一 花费10ms;任务二 花费 11ms;任务三 花费9ms;汇总任务花费1ms

串行执行:总花费时间是10+11+9+1 = 31ms多线程并行执行:如果是多核cpu 总花费时间是11+1=12ms

共享变量的内存可见性问题

内存模型

如上图，在Java内存模型中，分为主内存和工作内存,当线程需要对共享变量进行读写操作时，需要将主内存中的变量拷贝到工作内存中一份，后面的读写就在工作内存中进行读写

在CPU架构系统中，每一核的架构设计为控制器，运算器和缓存器，其中控制器包含一组寄存器和操作控制器,运算器执行算术逻辑运算。每一核都有自己的缓存器。

问题描述

1.线程A首次要获取M变量，在自己的缓存器中查找没有，从主内存中获取M变量并且拷贝到工作内存中，在工作内存中对M变量进行更改为1，并且在缓存器中缓存，将变量刷新到主内存中为1

2.线程B也要首次获取变量M。也在自己的缓存器中查找没有，从主内存中获取M变量为1拷贝到工作内存中，将M变量改为2，在缓存中缓存，将主内存中变量刷新为2

3.线程A再次获取M变量，从自己缓存中获取了M为1的值，这是造成了数据不一致问题，也就是说B对A的修改不可见

CAS操作

CAS（比较并交换）采用了一种乐观锁的思想，并且不断进行重试，是JDK中提供的非阻塞原子性操作

以compareAndSwapLong方法为例进行介绍

boolean compareAndSwapLong(Object obj,int valueOffset,long expect,long update)

CAS有四个操作数，分别是对象内存的位置，对象中变量的偏移量，变量预期值，和更新值

操作含义：如果对象obj中内存偏移量为valueOffset为expect时，就将update更新为expect，否则就会一直重试

先从内存中将数据取出，根据内存偏移量和变量的偏移量来取出取出来的值和数据的期望值进行比较，如果一直，数据更新 否则返回false

CAS在Unsafe类只有一次操作，如果需要不断重试，需要自己手动操作

存在问题：

ABA问题，竞争激烈情况下CPU飙升，占用率高只能使一个变量时原子性操作，不能时代码块是原子性操作

指令重排序

同一个线程内,JVM会在不影响正确性的前提下,可以调整语句的执行顺序

伪共享

为了解决cpu与内存之间的运行速度差异问题，添加了多级高速缓冲存储器，这些cache一般是集成到cpu内部，在cache内部是由行存储的，每一行成为cache行，cache行是与主内存进行数据交换的单位。当cpu访问一个变量时，如果cache中没有，就去主内存中查找数据，把该数据所在一个cache行大小的数据内存复制到cache中。

把多个变量放到cache中。当多个线程同时修改一个缓存行中的多个数据时，此时只允许一个线程操作缓存行，所以相比将每一个变量放入到一个缓存行中，性能会有所下降，这就是伪共享

避免方式：对象填充

锁

乐观锁和悲观锁

悲观锁：当一个线程持有一个共享变量时，认为其他线程一定会改动这个数据，所以在对数据处理时，先对其进行加锁，在整个数据的处理过程中，使数据一直处于锁定状态。类似于我们平时所使用的synchronized和lock锁。

乐观锁：乐观锁是相对与悲观锁来说的，他认为其他线程不会更改变量值，只是在更新数据的时候进行检测，比如CAS

公平锁和非公平锁

在ReentrantLock锁支持公平锁和非公平锁，但是在实现方式上只有两个方法不一样，一个是lock方法，另一个是tryacquire方法

ReentrantLock中默认的构造方法是非公平锁，在没有公平性的需求下尽量使用非公平锁，因为公平锁会带来性能消耗

源码解释：在lock方法中，非公平锁比公平锁多了一个cas判断，表示在获取锁之前，先用cas操作判断state属性进行判断当前锁是否还被占取，如果没被占取，直接获取锁，否则就和公平锁一样

static final class FairSync extends Sync {//......final void lock() {acquire(1);}//......}static final class NonfairSync extends Sync {//......final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}//......}

tryAcquire方法，在获取锁的时候，公平锁比非公平锁多了一个方法判断，判断当前阻塞队列是否有线程排队，如果有进入排队，如果队列不存在或没有元素直接获取锁

//非公平锁final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}//......}//公平锁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;}}//......}

独占锁和共享锁

锁可以被一个线程持有还是可以被多个线程共同持有，分为独占锁和共享锁

独占锁:保证任何时候只会一个线程持有，ReentrantLock就是以独占锁的形式实现的

共享锁:可以保证多个线程持有一把锁，ReadWriteLock就是一把读写锁，允许一个资源可以被多线程同时进行读操作

可重入锁

当一个线程要获取被其他线程持有的独占锁时，该线程会被阻塞，如果一个线程还要再继续获取被自己持有的独占锁时，是不会被阻塞的，也就是说该锁可重入，可以无限次的进入该锁被锁住的代码块。

synchronized是可重入锁，底层原理是在锁内部维护一个线程标识，用来标识锁是被哪个锁占用，然后关联一个计数器。一开始计数器的值为0，表示没有被任何线程占用，每当一个线程获取锁，计数器+1，释放锁后-1，当计数器的值为0，线程标识也会置为null，阻塞的线程也会竞争该锁

自旋锁

如果线程挂起，没有了cpu的执行权，就需要从用户态转为内核态被挂起。当线程获取锁的时候就需要从内核态转化为用户态去执行程序，如果高并发的场景下，这是很消耗性能的。自旋锁则是当前线程在获取锁的时候，如果没有获取到，不会马上被挂起阻塞，而是不放弃cpu的执行权，重试的获取锁（一般情况下是重试十次左右），很有可能在重试的过程中获取了锁资源。

高级部分

ThreadLocalRandom类

原子操作类

AtomicLong

JUC并发包下包含AtomicLong，AtomicInteger，AtomicBoolean等原子操作类，笔记主要以AtomicLong为讲解，其内部使用unsafe类实现。

主要函数

1.递增和递减

//通过unsafe方法，原子性的设置value值=初始值+1，返回递增后的值public final long incrementAndGet() {return unsafe.getAndAddLong(this, valueOffset, 1L) + 1L;}//通过unsafe方法，原子性的设置value值=原始值-1，返回递减后的值public final long decrementAndGet() {return unsafe.getAndAddLong(this, valueOffset, -1L) - 1L;}//返回值为原始值public final long getAndIncrement() {return unsafe.getAndAddLong(this, valueOffset, 1L);}//返回源氏值public final long getAndDecrement() {return unsafe.getAndAddLong(this, valueOffset, -1L);}

上面四个函数都是调用getAndAddLong方法实现，这个函数是原子性操作，第一个参数是当前实例对象的地址，第二个参数是value的偏移量，第三个参数是要设置的值

public final long getAndAddLong(Object var1, long var2, long var4) {long var6;//原始值do {var6 = this.getLongVolatile(var1, var2);} while(!pareAndSwapLong(var1, var2, var6, var6 + var4));return var6;}

2.比较并更新

如果期望值与原始值相同，就将原始值更新为update值并返回

public final boolean compareAndSet(long expect, long update) {return pareAndSwapLong(this, valueOffset, expect, update);}

缺点：在并发量较高的环境下，只允许一个线程对变量值进行更改，其他线程cas操作失败后，会无限的进行自旋尝试cas操作，cpu飙升，白白浪费cpu资源。

解决方式：java中提供了一个更好的类LongAdder，用来克服高并发下AtomicLong的缺点，在longadder中将变量分解为多个变量，让其他线程去竞争多个资源，就类似与火车站购票，如果只有一个窗口台，所有人只去这个窗口去排队买票，现在为了并发更高，开辟多个窗口，但是总票数不变，将这些分散到个个窗口。

result = base + cell[0]+…+cell[n]

LongAdder

Synchronized专题

Java并发包中锁原理

LockSupport工具类

主要用于挂起线程和唤醒线程，是创建锁和其他同步类的基础。

wait-notify：必须要放到同步代码中，sync中 还有一个就是notify必须放在wait后面，否则线程一直阻塞，无法唤醒。

lock-condition：放在lock中使用 先await然后再唤醒signal。

在LockSupport中就解决这两个缺点问题，主要方法有两个，park和unpark。LockSupport和每个使用它的线程都有一个凭证关联。每个线程都有一个相关的凭证，最多且有一个。

park方法：如果调用park方法的线程已经拿到许可证，则调用LockSupport.park()会直接返回，否则调用线程会被禁止参与线程调度，会被挂起阻塞。在其他线程调用调用unpark（Thread thread）方法将阻塞的线程作为参数，调用park方法被阻塞的会被唤醒。

unpark方法：为指定线程发送许可证，但是许可证最多只能有一个，累计无效。

可以突破wait/notify的原有调用顺序

因为unpark会给线程发放一个许可证，之后再调用park方法，就可以名正言顺的消费该凭证，所以不会阻塞。

唤醒两次后阻塞两次，还会阻塞

因为调用多次唤醒只会发放一个凭证，所以说连续调用两次unpark和调用一次unpark效果一样，指挥增加一个凭证，调用两次park要消费两个凭证，所以不能放行。

抽象同步队列AQS

简介：AQS是java并发包下的一个基础类，在并发包中的地位相当于jvm在java中的地位，很多并发工具类都是基于AQS实现，比如ReentrantLock互斥锁，ReentranReadWriteLock读写锁，CountDownLatch计数器，线程池中work锁机制都是基于AQS实现。AQS其实本身并不复杂，有一个volatile修饰并且基于cas修改的state属性，还有一个由node属性构成的一个双向链表，还有实现线程挂起以及唤醒的condition

CountDownLatch

代码演示

public class CounLatch {static CountDownLatch count = new CountDownLatch(2);public static void main(String[] args) throws InterruptedException {Thread thread1 = new Thread(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {count.countDown();}System.out.println("子线程一");});Thread thread2 = new Thread(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {count.countDown();}System.out.println("子线程二");});thread1.start();thread2.start();count.await();System.out.println("主线程");}}

上面代码中，创建了一个CountDownLatch对象，因为有两个子线程，所以传参为2.主线程调用 count.await();方法后会被阻塞。子线程执行完 count.countDown();后计数器会变为0.这时候主线程的await方法才会返回。

在实际项目开发中都不会直接操作线程，而是使用线程池来管理线程。所以这时候没办法使用join方法来实现等待线程执行终止，我们的countdownlatch就可以实现

package com.lsy;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/*** @description:* @projectName:com.lsy* @author:SpringLIU* @createTime:/09/01 23:31* @version:1.0*/public class CounLatch {static CountDownLatch count = new CountDownLatch(2);public static void main(String[] args) throws InterruptedException {ExecutorService executorService = Executors.newFixedThreadPool(2);executorService.submit(new Runnable() {@Overridepublic void run() {try {System.out.println("子线程一");Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {count.countDown();}}});executorService.submit(new Runnable() {@Overridepublic void run() {try {System.out.println("子线程二");Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {count.countDown();}}});count.await();executorService.shutdown();System.out.println("主线程");}}输出：子线程一子线程二主线程

CountDownLatch与join方法的区别

调用一个子线程的join方法，该线程会一直被阻塞到子线程执行完毕

而CountDownLatch则使用计数器来允许子线程运行完毕或者在运行中递减计数，也就是说在子线程运行的任何时候await方法返回而不一定必须等到线程结束

在线程池中无法使用join，这是就是可以使用CountDownLatch

实现原理

底层是AQS实现的，state属性表示计数器的值

构造方法

public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");this.sync = new Sync(count);}Sync(int count) {setState(count);}

await方法

委托了内部类sync去处理

public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);}

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {//判断当前线程是否中断，则抛出异常if (Thread.interrupted())throw new InterruptedException();//查看当前计数器是否为0，为0则直接返回，否则进入aqs队列中等待if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);}//判断当前state是否为0protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}

首先判断当前线程是否被中断，如果中断，抛出异常查看当前state属性是否为0，如果为0，程序正常执行，否则进去aqs队列中等待

countDown方法

该方法使state属性减一，当计数器的值递减为0则唤醒所有的阻塞队列中的线程，否则什么都不做

public void countDown() {//委托sync调用releaseShared方法sync.releaseShared(1);}

public final boolean releaseShared(int arg) //使state属性减一，并且判断state属性变为0if (tryReleaseShared(arg)) {//AQS释放锁doReleaseShared();return true;}return false;}

protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();//获取state属性if (c == 0)//如果刚开始c属性为0 直接返回return false;int nextc = c-1;//通过cas操作将state属性减一，然后进行判断if (compareAndSetState(c, nextc))return nextc == 0;}}}

ReentrantLock互斥锁源码逐行分析

一、提一嘴CAS

CAS可以保证在修改一个变量的时候是原子性的。

内存中有一个int类型的变量x。

基于CAS修改：

获取变量x的oldValue基于Unsafe类提供的compareAndSwap方法，将x从oldValue修改为newValue（原子性）

二、简单阐述AQS核心内容

ReentrantLock是基于AQS实现的

AQS提供了两个核心内容：

int类型的state属性（如果线程将state从0改为1（基于CAS），并且成功了，就代表获取锁资源成功）一个双向链表，存储一些没有获取到锁资源的线程信息。（两个线程竞争锁资源，线程A获取到锁了，此时线程B无法获取锁资源，去双向链表中排队）

三、查看lock方法（公平锁和非公平锁）

ReentrantLock可以实现公平锁和非公平锁。

分别是FairSync以及NonfairSync，默认情况会使用NonfairSync

// 非公平锁的lockfinal void lock() {// 直接基于CAS，尝试将state从0改为1if (compareAndSetState(0, 1))// 获取锁成功，需要将exclusiveOwnerThread属性设置为当前线程，代表当前线程拿到锁资源setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}// 公平锁的lockfinal void lock() {acquire(1);}

四、查看acquire方法

// AQS提供的acquire方法，公平和非公平都使用这个方法public final void acquire(int arg) {// 第一个操作：tryAcquire，尝试拿一次锁资源，没拿到返回false。// 第二个操作：addWaiter，拿锁失败，去排队。// 第三个操作：acquireQueued，可以获取锁资源失败的将线程挂起if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))// 省略部分代码}

4.1 执行流程

4.1.1 tryAcquire执行流程

4.1.2 addWaiter方法

4.1.3 acquireQueued方法流程

4.2 tryAcquire

4.2.1 非公平锁

// 非公平锁的tryAcquirefinal boolean nonfairTryAcquire(int acquires) {// 拿到当前线程。final Thread current = Thread.currentThread();// 拿到state的原值int c = getState();// 如果state为0，说明没有线程持有锁资源if (c == 0) {// 基于CAS尝试将state从0改为1，如果成功，返回trueif (compareAndSetState(0, acquires)) {// 讲持有锁的线程设置为当前线程setExclusiveOwnerThread(current);// 返回truereturn true;}}// 如果state不为0，查看是否是锁重入的操作// c肯定是大于0的，有线程持有锁。// 持有锁的线程是不是我自己else if (current == getExclusiveOwnerThread()) {// 将state + 1得到新值int nextc = c + acquires;// 判断锁重入次数是否已经到达到了最大值。if (nextc < 0)throw new Error("Maximum lock count exceeded");// 011111111 11111111 11111111 11111111// 100000000 00000000 00000000 00000000// 将新值set给state。setState(nextc);// 返回true，锁重入成功，也属于获取锁成功return true;}// 拿锁失败。return false;}

4.2.2 公平锁

没有线程在AQS的双向链表排队时，可以执行CAS尝试拿锁有线程在AQS的双向链表排队时，如果当前线程排在head.next位置，可以执行CAS尝试拿锁

// 公平锁的实现// 公平锁和非公平锁的tryAcquire方法的实现，就差一个hasQueuedPredecessors方法// 如果当前锁没有线程持有，非公平锁直接抢，公平锁先看下有没有排队的protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {// 只查看当前判断// 只有返回false的时候，才会尝试竞争锁资源if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0)throw new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}// 查看是否有排队的线程// 想看懂这，先掌握一个AQS的双向链表public final boolean hasQueuedPredecessors() {Node t = tail; Node h = head;Node s;// head和tail是否相等// 如果相等，没有排队的线程，直接返回false，公平锁就可以执行CAS尝试拿锁// 如果不相等，有排队的线程return h != t &&// 排在第一位的线程，是允许执行CAS尝试锁资源的// head.next就是排在第一位的线程// 只是为了拿到head.next，并且做非空判断// 有排队线程，并且排队的线程是自己。((s = h.next) == null || s.thread != Thread.currentThread());}

4.3 addWaiter

// AQS提供的addWaiter// 线程获取锁资源失败，会执行addWaiter，去排队。private Node addWaiter(Node mode) {// 将当前线程封装为Node对象，这个mode，代表是互斥锁。Node node = new Node(Thread.currentThread(), mode);// 拿到了tail节点Node pred = tail;// 如果tail节点不为null，AQS链表不为空，直接将当前节点插入进去。if (pred != null) {// 图中说的1，2，3的插入流程node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}// 查看enq就可以看到上述的操作流程enq(node);return node;}// 添加Node到AQS链表private Node enq(final Node node) {// 死循环。一定要将当前线程的Node放到AQS的双向链表中。for (;;) {// 获取tail节点Node t = tail;// 如果tail为null，AQS没有排队的节点if (t == null) {// Must initialize 必须初始化// new一个Node对象，基于CAS方式将head指向当前Nodeif (compareAndSetHead(new Node()))// 将tail指向headtail = head;} // 说明AQS中已经有至少一个节点了，else {//将当前Node的prev指向tailnode.prev = t;// 基于CAS的方式，将之前tail替换为当前Nodeif (compareAndSetTail(t, node)) {// 将之前的tail的next指向当前Nodet.next = node;// 返回当前Nodereturn t;}}}}

4.4 acquireQueued

// 尝试挂起线程和尝试竞争锁的acquireQueuedfinal boolean acquireQueued(final Node node, int arg) {// 拿锁失败的标记：默认为trueboolean failed = true;// 死循环。for (;;) {// 拿到当前node的上一个节点pfinal Node p = node.predecessor();// 如果上一个节点是head，当前node可以执行tryAcquire尝试竞争锁资源。// 如果tryAcquire返回true，进去if代码块，代表拿锁成功if (p == head && tryAcquire(arg)) {// ~~~这里暂时先不看~~~~setHead(node);p.next = null; // help GCfailed = false;return false;}// 执行的这，要么没有竞争锁的资格，要么没抢到锁。if (shouldParkAfterFailedAcquire(p, node) &&// 到这说明，prev节点的状态已经为-1了，可以执行parkAndCheckInterrupt方法挂起线程// 内部执行了LockSupport.park(this);最终调用了unsafe的park方法parkAndCheckInterrupt())}}private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// 获取上一个节点的状态，wsint ws = pred.waitStatus;// 如果上一个几点状态为-1，直接返回true，代表可以挂起if (ws == -1)return true;// 判断上一个节点的状态是不是取消状态，（取消状态为1）if (ws > 0) {// 往前找到一个节点状态不为1的Node，跟在后面do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {// 只要上一个节点状态不是取消状态，就将其设置为-1compareAndSetWaitStatus(pred, ws, -1);}return false;}

五、查看unlock方法

unlock不分为公平锁和非公平锁，都执行一个unlock方法

// 释放锁资源public void unlock() {sync.release(1);}// unlock中执行的释放锁public final boolean release(int arg) {// 释放锁资源。if (tryRelease(arg)) {// 锁资源释放干净了// 拿到headNode h = head;// h != null是健壮性判断// head节点只可能出现两种状态，要么是0，要么是-1if (h != null && h.waitStatus != 0)// 这里head节点状态为-1，就代表后继节点可能线程挂起了。、// 走unpark方法，去唤醒后继节点。// 如果没有哨兵节点的状态做判断，不知道是否需要在释放锁资源之后唤醒后继节点unparkSuccessor(h);return true;}// 锁资源没释放干净。return false;}// 尝试释放锁资源protected final boolean tryRelease(int releases) {// 获取state，然后 - 1等到cint c = getState() - releases;// 释放锁资源的必须是持有锁资源的线程。if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();// 声明了变量free，代表锁资源是否释放干净了。boolean free = false;// 如果c == 0，代表锁资源释放干净了。if (c == 0) {// 设置为true，free = true;// 并且将持有锁线程的标识设置为nullsetExclusiveOwnerThread(null);}// 设置给statesetState(c);// 返回freereturn free;}// 唤醒线程的操作。node是head节点private void unparkSuccessor(Node node) {// 拿到head节点的状态int ws = node.waitStatus;// 如果状态为-1if (ws < 0)// 已经要执行唤醒操作了，将状态改为0compareAndSetWaitStatus(node, ws, 0);// ==================================// 拿到head的next节点Node s = node.next;// 如果s为null，或者s的状态是取消if (s == null || s.waitStatus > 0) {// 到这，说明s不能唤醒了// 先设置为nulls = null;// 遍历整个双向链表，找到离head最新的有效节点// 为什么从后往前遍历，找离head最近的。 明显从前往后找块啊。// addWaiter方法中插入节点的方式，导致从前往后找，可能会丢失节点。// 为了保证结构正确，需要从后往前找。for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}// 如果最后找到的节点，不为null，执行一波唤醒操作if (s != null)// 调用Unsafe的唤醒操作LockSupport.unpark(s.thread);}

ThreadPoolExecutor源码分析

JDK自带的线程池

单例线程池

核心线程为1个，最大线程数也是1个，当前线程池只有一个线程，阻塞队列可以无限的接受任务

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}

FinalizableDelegatedExecutorService的作用？

static class FinalizableDelegatedExecutorServiceextends DelegatedExecutorService {FinalizableDelegatedExecutorService(ExecutorService executor) {super(executor);}protected void finalize() {super.shutdown();}}

重写了finalize方法，finalize的作用？

每一个对象只要可达性分析无法引用到，就需要被垃圾回收器回收掉，这个对象重写了finalize，在回收之前会执行finalize方法。

为什么线程池不用之后要执行shutdown？

结束掉还没有结束的线程，如果线程池使用完毕后，线程池虽然被干掉了，但是线程依然存在。比如在项目中，临时使用线程池解决某一个场景，当场景结束后，线程池使用结束，但是核心线程一直存在，这时就需要shutdown解决核心线程。

可以基于finalize保证线程池被销毁前，一定会持续shutdown方法吗？

不可以保证，因为finalize方法执行时会创建一个守护线程，他不保证finalize方法执行结束，

定长线程池

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}

相对于单例线程池只指定了核心工作线程的个数

缓存线程池

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}

当任务提交到线程池中，核心线程池中的个数为0，所以会被放到阻塞队列中，会被创建一个非核心工作线程去处理任务。

定时任务线程池

本质还是ThreadPoolExecutor，但是阻塞队列还是采用DelayedQueue，比如为了周期性的执行任务，将执行完毕的任务再仍回到阻塞队列

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);}

工作窃取线程池

public static ExecutorService newWorkStealingPool() {return new ForkJoinPool(Runtime.getRuntime().availableProcessors(),ForkJoinPool.defaultForkJoinWorkerThreadFactory,null, true);}

使用了ForkJoinPool，这个线程池为了大任务存在，一个拆分一个聚合。比如一个任务需要执行五分钟，如果我把这个任务拆分成5块，交给5个线程。理论上1分钟就可以完成。

若公司设计到异步编程，可以采用CompletableFutrue时默认常采用的线程池就是ForkJoinPool。

为什么要使用线程池？

减少开销，便于管理。

一、线程池的核心属性

为了可以更清晰的去分析线程池的源码，核心属性必须掌握

ctl属性，维护这线程池状态以及工作线程个数。

//线程池的核心属性// 核心属性就是ctl，如果不认识AtomicInteger，就把ctl当成int看// ctl一个int类型表示了线程池的2大核心内容// 第一个：线程池的状态// 第二个：工作线程个数// ctl是int类型，而int是32个bit位组成的数值// 高3位记录：线程池的状态，低29位记录：工作线程个数private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// 这个29就是为了方便对int类型数值进程分割。private static final int COUNT_BITS = 29;// 工作线程最大个数private static final int CAPACITY = (1 << COUNT_BITS) - 1;00000000 00000000 00000000 0000000100100000 00000000 00000000 0000000000011111 11111111 11111111 11111111// 高3位记录的线程池状态// RUNNING：属于正常状态，可以正常接收任务，并且处理任务private static final int RUNNING = -1 << COUNT_BITS;// SHUTDOWN：处理关闭状态，不接收新任务，但是会处理完线程池中之前接收到的任务private static final int SHUTDOWN = 0 << COUNT_BITS;// STOP：处理停止状态，不接收新任务，中断正在执行的任务，之前接收的任务也不去处理private static final int STOP = 1 << COUNT_BITS;// TIDYING：过渡状态，是从SHUTDOWN或者STOP转换过来的，工作线程都凉凉了，任务也处理完了（了解）private static final int TIDYING = 2 << COUNT_BITS;// TERMINATED：终止状态，从TIDYING执行了terminated方法，转换到TERMINATED（了解）private static final int TERMINATED = 3 << COUNT_BITS;// 线程池最大允许有多少个工作线程？

二、线程池的7个参数

JUC包下的Executors提供了几种JDK自带的线程池构建方式，可以不用手动去构建。

but，不推荐使用上述方式去构建线程池。

为了让咱们可以更好的去控制线程池对象，推荐直接采用new的方式去构建ThreadPoolExecutor

为了去手动new，就要对ThreadPoolExecutor对象的有参构造参数有掌握

// 线程池的7个参数// 核心线程数，默认不会被干掉public ThreadPoolExecutor(int corePoolSize, // 最大线程数，在核心线程数之外的非核心线程个数，// maximumPoolSize - corePoolSize = 非核心线程个数int maximumPoolSize, // 最大空闲时间（一般针对非核心线程） long keepAliveTime, // 时间单位（一般针对非核心线程） TimeUnit unit, // 工作队列，核心线程个数满足corePoolSize，再来任务，就扔到工作队列 BlockingQueue<Runnable> workQueue, // 线程工厂，为了更方面后期出现故障时，定位问题，在构建线程时，指定好// 一些细节信息，给个名字~~ThreadFactory threadFactory, // 拒绝策略，核心满了，队列满了，非核心到位了，再来任务走拒绝策略RejectedExecutionHandler handler) {// 省略部分代码}// 在线程池中，核心线程和非核心线程都属于工作线程。new ThreadPoolExecutor(5,7,1,TimeUnit.SECOND,new ArrayBlockingQueue<>(10),线程工厂,拒绝策略)

三、线程池的执行流程（execute方法）

要将任务提交给线程池执行时，需要调用线程池的execute方法，将任务传递进去

// 线程池执行任务的核心方法// 线程池的执行流程图就是基于这个方法画出来的。public void execute(Runnable command) {// 非空判断~if (command == null) throw new NullPointerException();// 获取ctl属性，命名cint c = ctl.get();// =======================创建核心线程=============================== // workerCountOf(c)：获取工作线程的个数// 如果工作线程数 小于 核心线程数if (workerCountOf(c) < corePoolSize) {// 通过addWorker方法创建 核心线程。// command是传递的任务，true代表核心线程// 如果创建工作线程成功：返回true// 如果创建工作线程失败：返回falseif (addWorker(command, true))// 告辞，任务交给线程执行了return;// 到这，说明失败了，走下一流程c = ctl.get();}// =======================将任务添加到工作队列=============================== // 判断线程池状态是不是RUNNING// 只有状态正常，才会走offer方法（添加任务到工作队列）// offer方法，添加成功返回true，添加失败返回falseif (isRunning(c) && workQueue.offer(command)) {// 任务已经放到工作队列了。// 任务放到队列后，要重新检查一次，重新拿到ctlint recheck = ctl.get();// 判断线程池状态是不是RUNNING// 如果状态不是RUNNING，将刚刚放进来的任务从队列移除~~~if (!isRunning(recheck) && remove(command))// 执行拒绝策略。 reject(command);// 如果状态是RUNNING，同时工作线程数是0个。else if (workerCountOf(recheck) == 0)// 如果没有工作线程，添加一个非核心,空任务线程去解决掉工作队列没人处理的任务addWorker(null, false);}// =======================创建非核心线程=============================== // addWorker创建非核心线程，如果成功，返回true// 如果添加非核心线程失败，执行reject方法else if (!addWorker(command, false))// =======================拒绝策略=============================== // 执行拒绝策略 reject(command);}// 达尔优，天空轴~~~

四、线程池如何添加工作线程（addWorker方法）

前面就提到了添加工作线程，查看addWorker的源码

// 添加工作线程的方式private boolean addWorker(Runnable firstTask, boolean core) {// 判断是否可以正常添加工作线程retry:for (;;) {//=======================判断线程池状态=================================== // 获取ctl属性int c = ctl.get();// 获取线程状态int rs = runStateOf(c);// 如果线程池状态不是RUNNING（正常到这，就不能添加工作线程了）if (rs >= SHUTDOWN &&// 可能出现工作队列有任务，但是线程池没有工作线程// 需要添加一个非核心空任务的工作线程，这里就是。// 只要一个不满足，就打破了之前效果，不需要添加!(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))// 当前状态不允许添加工作线程，返回falsereturn false;for (;;) {// 内部有点泄漏，马上溢出，释放一下内存//=======================判断工作线程个数=================================== // 阿巴阿巴~~int wc = workerCountOf(c);// 如果工作线程个数，大于线程池允许的工作线程最大数，满足，告辞，不能添加if (wc >= CAPACITY ||// core：true添加核心，false添加非核心// 根据不同情况，比较不同的参数// 如果线程数不满足new线程池时的参数，告辞，不能添加wc >= (core ? corePoolSize : maximumPoolSize))// 不能添加。return false;// 没进到if，可以添加工作线程// 基于CAS的方式，对ctl进行 + 1操作// 线程A,线程Bif (compareAndIncrementWorkerCount(c))// 如果CAS成功，直接跳出外层循环，执行添加工作线程，并且启动工作线程break retry;// 重新获取ctlc = ctl.get(); // 如果线程池状态不变，直接正常重新执行内部循环if (runStateOf(c) != rs)// 如果线程池状态变了，重新判断线程池状态，走外部循环。continue retry;}}// 真正的去添加工作线程，并且启动工作线程// 声明了三个标识// workerStarted：工作线程启动了吗？boolean workerStarted = false;// workerAdded：工作线程创建了吗？boolean workerAdded = false;// Worker：工作线程Worker w = null;try {// 创建工作线程w = new Worker(firstTask);// 获取到工作线程中的Threadfinal Thread t = w.thread;// 这个if几乎不会发生，除非你写ThreadFactory有问题，或者是业务没通过if (t != null) {try {// 拿到线程池状态int rs = runStateOf(ctl.get());// rs < SHUTDOWN：满足，代表线程池状态ok是RUNNINGif (rs < SHUTDOWN ||// 状态是SHUTDOWN，任务是空，阿巴阿巴~~~(rs == SHUTDOWN && firstTask == null)) {// 除非你写ThreadFactory有问题，或者是业务没通过，线程运行了，这里就扔异常if (t.isAlive()) throw new IllegalThreadStateException();// 添加Worker工作线程到workers。// private final HashSet<Worker> workers = new HashSet<Worker>();workers.add(w);// 在记录工作线程的历史最大值int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;// 工作线程创建了！！！workerAdded = true;}}// 添加工作线程成功，那就启动线程if (workerAdded) {t.start();// 工作线程启动了！！！workerStarted = true;}}} finally {// 工作线程启动失败if (!workerStarted)// 将Worker从HashSet移除，并且对ctl - 1addWorkerFailed(w);}return workerStarted;}

五、工作线程执行任务方式（runWorker方法）

runWorker其实就是启动工作线程做的事。

核心就是Worker对象在调用了start方法后，会执行Worker类中的run方法。

在run方法中，执行了runWorker方法

// 工作线程干活的方法final void runWorker(Worker w) {// 获取当前线程Thread wt = Thread.currentThread();// 第一次从Worker中拿任务，赋值给task属性Runnable task = w.firstTask;// 将Worker中的任务置位空w.firstTask = null;try {// task != null：说明任务是最开始addWorker传递过来的。// (task = getTask()) != null：说明任务是从工作队列中获取到的while (task != null || (task = getTask()) != null) {// 删掉了部分catch~~~try {// 前置增强（勾子函数）beforeExecute(wt, task);try {// 执行任务task.run();} finally {// 后置增强（勾子函数）afterExecute(task, thrown);}} finally {// 任务值为空task = null;// 记录当前Worker完成了一个任务pletedTasks++;}}}}

六、工作线程获取任务方式&结束工作线程方式（getTask方法）

Worker工作线程除了addWorker自带的任务之外，就是从工作队列获取

查看getTask方法做了什么事

// 工作线程从工作队列获取任务private Runnable getTask() {// 超时了咩？ 默认没！boolean timedOut = false; // 死循环。for (;;) {// 拿到ctl，获取线程池状态int c = ctl.get();int rs = runStateOf(c);// 第一个：线程池状态是SHUTDOWN，并且阻塞队列为空// 第二个：线程池是STOP状态if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {// ctl - 1，当前线程可以销毁了~decrementWorkerCount();return null;}// 线程池状态正常。// 重新获取工作线程个数int wc = workerCountOf(c);// 回头看！// allowCoreThreadTimeOut：默认为false，如果为true，核心线程也要超时// wc > corePoolSize：工作线程大于核心线程数（现在有非核心线程）boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 如果工作线程大于最大线程数，基本不会满足，就是false// (timed && timedOut)：true代表之前走了poll，但是没拿到任务if ((wc > maximumPoolSize || (timed && timedOut))// 工作线程至少2个// 工作队列为空&& (wc > 1 || workQueue.isEmpty())) {// 干掉一个非核心线程。（循环结束拿不到任务即可）// CAS对ctl - 1，销毁线程if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 基于阻塞队列的poll或者是take获取任务// poll可以指定等待多久拿任务（非核心线程）// take死等任务。（核心线程）Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();// 如果拿到任务，正常返回if (r != null)return r;// 没拿到任务，timeOut是truetimedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

一般线程池参数设置多少？

这个没有固定的公式，虽然很多书上有设置公式，但是不一定适合你的业务。

如果要搞一个合适的线程池参数设置，你需要去动态的监控线程池，并且可以动态修改。

只能根据细粒度的测试慢慢调试一个比较合适的参数。

线程池提供了核心属性的get方法，还有核心参数动态set的功能。

自带的拒绝策略？

都是工作线程，到时间，怎么知道这个被销毁，那个不销毁？

并发集合

ConcurrentHashMap

一、存储结构

ConcurrentHashMap是线程安全的HashMap

ConcurrentHashMap在JDK1.8中是以CAS+synchronized实现的线程安全

CAS：在没有hash冲突时（Node要放在数组上时）

synchronized：在出现hash冲突时（Node存放的位置已经有数据了）

存储的结构：数组+链表+红黑树

二、存储操作

2.1 put方法

public V put(K key, V value) {// 在调用put方法时，会调用putVal，第三个参数默认传递为false// 在调用putIfAbsent时，会调用putVal方法，第三个参数传递的为true// 如果传递为false，代表key一致时，直接覆盖数据// 如果传递为true，代表key一致时，什么都不做，key不存在，正常添加（Redis，setnx）return putVal(key, value, false);}

2.2 putVal方法-散列算法

final V putVal(K key, V value, boolean onlyIfAbsent) {// ConcurrentHashMap不允许key或者value出现为null的值，跟HashMap的区别if (key == null || value == null) throw new NullPointerException();// 根据key的hashCode计算出一个hash值，后期得出当前key-value要存储在哪个数组索引位置int hash = spread(key.hashCode());// 一个标识，在后面有用！int binCount = 0;// 省略大量的代码……}// 计算当前Node的hash值的方法static final int spread(int h) {// 将key的hashCode值的高低16位进行^运算，最终又与HASH_BITS进行了&运算// 将高位的hash也参与到计算索引位置的运算当中// 为什么HashMap、ConcurrentHashMap，都要求数组长度为2^n// HASH_BITS让hash值的最高位符号位肯定为0，代表当前hash值默认情况下一定是正数，因为hash值为负数时，有特殊的含义// static final int MOVED= -1; // 代表当前hash位置的数据正在扩容！// static final int TREEBIN = -2; // 代表当前hash位置下挂载的是一个红黑树// static final int RESERVED = -3; // 预留当前索引位置……return (h ^ (h >>> 16)) & HASH_BITS;// 计算数组放到哪个索引位置的方法 (f = tabAt(tab, i = (n - 1) & hash)// n：是数组的长度}00001101 00001101 00101111 10001111 - h = key.hashCode运算方式00000000 00000000 00000000 00001111 - 15 (n - 1)&((00001101 00001101 00101111 10001111 - h^00000000 00000000 00001101 00001101 - h >>> 16)&01111111 11111111 11111111 11111111 - HASH_BITS)

2.3 putVal方法-添加数据到数组&初始化数组

final V putVal(K key, V value, boolean onlyIfAbsent) {// 省略部分代码…………// 将Map的数组赋值给tab，死循环for (Node<K,V>[] tab = 0;;) {// 声明了一堆变量~~// n:数组长度// i:当前Node需要存放的索引位置// f: 当前数组i索引位置的Node对象// fn：当前数组i索引位置上数据的hash值Node<K,V> f; int n, i, fh;// 判断当前数组是否还没有初始化if (tab == null || (n = tab.length) == 0)// 将数组进行初始化。tab = initTable();// 基于 (n - 1) & hash 计算出当前Node需要存放在哪个索引位置// 基于tabAt获取到i位置的数据else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 现在数组的i位置上没有数据，基于CAS的方式将数据存在i位置上if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))// 如果成功，执行break跳出循环，插入数据成功break; }// 判断当前位置数据是否正在扩容……else if ((fh = f.hash) == MOVED)// 让当前插入数据的线程协助扩容tab = helpTransfer(tab, f);// 省略部分代码…………}// 省略部分代码…………}sizeCtl：是数组在初始化和扩容操作时的一个控制变量-1：代表当前数组正在初始化小于-1：低16位代表当前数组正在扩容的线程个数（如果1个线程扩容，值为-2，如果2个线程扩容，值为-3）0：代表数据还没初始化大于0：代表当前数组的扩容阈值，或者是当前数组的初始化大小// 初始化数组方法private final Node<K,V>[] initTable() {// 声明标识Node<K,V>[] tab; int sc;// 再次判断数组没有初始化，并且完成tab的赋值while ((tab = table) == null || tab.length == 0) {// 将sizeCtl赋值给sc变量，并判断是否小于0if ((sc = sizeCtl) < 0)Thread.yield(); // 可以尝试初始化数组，线程会以CAS的方式，将sizeCtl修改为-1，代表当前线程可以初始化数组else if (pareAndSwapInt(this, SIZECTL, sc, -1)) {// 尝试初始化！try {// 再次判断当前数组是否已经初始化完毕。if ((tab = table) == null || tab.length == 0) {// 开始初始化，// 如果sizeCtl > 0，就初始化sizeCtl长度的数组// 如果sizeCtl == 0，就初始化默认的长度int n = (sc > 0) ? sc : DEFAULT_CAPACITY;// 初始化数组！Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 将初始化的数组nt，赋值给tab和tabletable = tab = nt;// sc赋值为了数组长度 - 数组长度 右移 2位 16 - 4 = 12// 将sc赋值为下次扩容的阈值sc = n - (n >>> 2);}} finally {// 将赋值好的sc，设置给sizeCtlsizeCtl = sc;}break;}}return tab;}

2.4 putVal方法-添加数据到链表

final V putVal(K key, V value, boolean onlyIfAbsent) {// 省略部分代码…………int binCount = 0;for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;// n:数组长度// i:当前Node需要存放的索引位置// f: 当前数组i索引位置的Node对象// fn：当前数组i索引位置上数据的hash值// 省略部分代码…………else {// 声明变量为oldValV oldVal = null;// 基于当前索引位置的Node，作为锁对象……synchronized (f) {// 判断当前位置的数据还是之前的f么……（避免并发操作的安全问题）if (tabAt(tab, i) == f) {// 再次判断hash值是否大于0（不是树）if (fh >= 0) {// binCount设置为1（在链表情况下，记录链表长度的一个标识）binCount = 1;// 死循环，每循环一次，对binCountfor (Node<K,V> e = f;; ++binCount) {// 声明标识ekK ek;// 当前i索引位置的数据，是否和当前put的key的hash值一致if (e.hash == hash &&// 如果当前i索引位置数据的key和put的key == 返回为true// 或者equals相等((ek = e.key) == key || (ek != null && key.equals(ek)))) {// key一致，可能需要覆盖数据！// 当前i索引位置数据的value复制给oldValoldVal = e.val;// 如果传入的是false，代表key一致，覆盖value// 如果传入的是true，代表key一致，什么都不做！if (!onlyIfAbsent)// 覆盖valuee.val = value;break;}// 拿到当前指定的Node对象Node<K,V> pred = e;// 将e指向下一个Node对象,如果next指向的是一个null，可以挂在当前Node下面if ((e = e.next) == null) {// 将hash，key，value封装为Node对象，挂在pred的next上pred.next = new Node<K,V>(hash, key,value, null);break;}}}// 省略部分代码…………}}// binCount长度不为0if (binCount != 0) {// binCount是否大于8（链表长度是否 >= 8）if (binCount >= TREEIFY_THRESHOLD)// 尝试转为红黑树或者扩容// 基于treeifyBin方法和上面的if判断，可以得知链表想要转为红黑树，必须保证数组长度大于等于64，并且链表长度大于等于8// 如果数组长度没有达到64的话，会首先将数组扩容treeifyBin(tab, i);// 如果出现了数据覆盖的情况，if (oldVal != null)// 返回之前的值return oldVal;break;}}}// 省略部分代码…………}// 为什么链表长度为8转换为红黑树，不是能其他数值嘛？// 因为布松分布The main disadvantage of per-bin locks is that other update* operations on other nodes in a bin list protected by the same* lock can stall, for example when user equals() or mapping* functions take a long time. However, statistically, under* random hash codes, this is not a common problem. Ideally, the* frequency of nodes in bins follows a Poisson distribution* (/wiki/Poisson_distribution) with a* parameter of about 0.5 on average, given the resizing threshold* of 0.75, although with a large variance because of resizing* granularity. Ignoring variance, the expected occurrences of* list size k are (exp(-0.5) * pow(0.5, k) / factorial(k)). The* first values are:** 0: 0.60653066* 1: 0.30326533* 2: 0.07581633* 3: 0.01263606* 4: 0.00157952* 5: 0.00015795* 6: 0.00001316* 7: 0.00000094* 8: 0.00000006* more: less than 1 in ten million

三、扩容操作

3.1 treeifyBin方法触发扩容

// 在链表长度大于等于8时，尝试将链表转为红黑树private final void treeifyBin(Node<K,V>[] tab, int index) {Node<K,V> b; int n, sc;// 数组不能为空if (tab != null) {// 数组的长度n，是否小于64if ((n = tab.length) < MIN_TREEIFY_CAPACITY)// 如果数组长度小于64，不能将链表转为红黑树，先尝试扩容操作tryPresize(n << 1);// 省略部分代码……}}

3.2 tryPreSize方法-针对putAll的初始化操作

// size是将之前的数组长度 左移 1位得到的结果private final void tryPresize(int size) {// 如果扩容的长度达到了最大值，就使用最大值// 否则需要保证数组的长度为2的n次幂// 这块的操作，是为了初始化操作准备的，因为调用putAll方法时，也会触发tryPresize方法// 如果刚刚new的ConcurrentHashMap直接调用了putAll方法的话，会通过tryPresize方法进行初始化int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size + (size >>> 1) + 1);//保证数组大小为2的n次方// 这些代码和initTable一模一样// 声明scint sc;// 将sizeCtl的值赋值给sc，并判断是否大于0，这里代表没有初始化操作，也没有扩容操作sizeCtl：是数组在初始化和扩容操作时的一个控制变量-1：代表当前数组正在初始化小于-1：低16位代表当前数组正在扩容的线程个数（如果1个线程扩容，值为-2，如果2个线程扩容，值为-3）0：代表数据还没初始化大于0：代表当前数组的扩容阈值，或者是当前数组的初始化大小while ((sc = sizeCtl) >= 0) {// 将ConcurrentHashMap的table赋值给tab，并声明数组长度nNode<K,V>[] tab = table; int n;// 数组是否需要初始化if (tab == null || (n = tab.length) == 0) {// 进来执行初始化// sc是初始化长度，初始化长度如果比计算出来的c要大的话，直接使用sc，如果没有sc大，// 说明sc无法容纳下putAll中传入的map，使用更大的数组长度n = (sc > c) ? sc : c;// 设置sizeCtl为-1，代表初始化操作if (pareAndSwapInt(this, SIZECTL, sc, -1)) {try {// 再次判断数组的引用有没有变化if (table == tab) {// 初始化数组Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 数组赋值table = nt;// 计算扩容阈值sc = n - (n >>> 2);}} finally {// 最终赋值给sizeCtlsizeCtl = sc;}}}// 如果计算出来的长度c如果小于等于sc，直接退出循环结束方法// 数组长度大于等于最大长度了，直接退出循环结束方法else if (c <= sc || n >= MAXIMUM_CAPACITY)break;// 省略部分代码}}// 将c这个长度设置到最近的2的n次幂的值， 15 - 1617 - 32// c == size + (size >>> 1) + 1// size = 1700000000 00000000 00000000 00010001+ 00000000 00000000 00000000 00001000+00000000 00000000 00000000 00000001// c = 2600000000 00000000 00000000 00011010private static final int tableSizeFor(int c) {// 00000000 00000000 00000000 00011001int n = c - 1;// 00000000 00000000 00000000 00011001// 00000000 00000000 00000000 00001100// 00000000 00000000 00000000 00011101n |= n >>> 1;// 00000000 00000000 00000000 00011101// 00000000 00000000 00000000 00000111// 00000000 00000000 00000000 00011111n |= n >>> 2;// 00000000 00000000 00000000 00011111// 00000000 00000000 00000000 00000001// 00000000 00000000 00000000 00011111n |= n >>> 4;// 00000000 00000000 00000000 00011111// 00000000 00000000 00000000 00000000// 00000000 00000000 00000000 00011111n |= n >>> 8;// 00000000 00000000 00000000 00011111n |= n >>> 16;// 00000000 00000000 00000000 00100000return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;}

3.3 tryPreSize方法-计算扩容戳并且查看BUG

private final void tryPresize(int size) {// n：数组长度while ((sc = sizeCtl) >= 0) {// 判断当前的tab是否和table一致，else if (tab == table) {// 计算扩容表示戳，根据当前数组的长度计算一个16位的扩容戳// 第一个作用是为了保证后面的sizeCtl赋值时，保证sizeCtl为小于-1的负数/T// 第二个作用用来记录当前是从什么长度开始扩容的int rs = resizeStamp(n);// BUG --- sc < 0，永远进不去~// 如果sc小于0，代表有线程正在扩容。if (sc < 0) {// 省略部分代码……协助扩容的代码（进不来~~~~）}// 代表没有线程正在扩容，我是第一个扩容的。else if (pareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))// 省略部分代码……第一个扩容的线程……}}}// 计算扩容表示戳// 32 = 00000000 00000000 00000000 00100000// Integer.numberOfLeadingZeros(32) = 26// 1 << (RESIZE_STAMP_BITS - 1) // 00000000 00000000 10000000 00000000// 00000000 00000000 00000000 00011010// 00000000 00000000 10000000 00011010static final int resizeStamp(int n) {return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));}

3.4 tryPreSize方法-对sizeCtl的修改以及条件判断的BUG

private final void tryPresize(int size) {// sc默认为sizeCtlwhile ((sc = sizeCtl) >= 0) {else if (tab == table) {// rs：扩容戳 00000000 00000000 10000000 00011010int rs = resizeStamp(n);if (sc < 0) {// 说明有线程正在扩容，过来帮助扩容Node<K,V>[] nt;// 依然有BUG// 当前线程扩容时，老数组长度是否和我当前线程扩容时的老数组长度一致// 00000000 00000000 10000000 00011010if ((sc >>> RESIZE_STAMP_SHIFT) != rs // 10000000 00011010 00000000 00000010 // 00000000 00000000 10000000 00011010// 这两个判断都是有问题的，核心问题就应该先将rs左移16位，再追加当前值。// 这两个判断是BUG// 判断当前扩容是否已经即将结束|| sc == rs + 1 // sc == rs << 16 + 1 BUG// 判断当前扩容的线程是否达到了最大限度|| sc == rs + MAX_RESIZERS // sc == rs << 16 + MAX_RESIZERS BUG// 扩容已经结束了。|| (nt = nextTable) == null // 记录迁移的索引位置，从高位往低位迁移，也代表扩容即将结束。|| transferIndex <= 0)break;// 如果线程需要协助扩容，首先就是对sizeCtl进行+1操作，代表当前要进来一个线程协助扩容if (pareAndSwapInt(this, SIZECTL, sc, sc + 1))// 上面的判断没进去的话，nt就代表新数组transfer(tab, nt);}// 是第一个来扩容的线程// 基于CAS将sizeCtl修改为 10000000 00011010 00000000 00000010 // 将扩容戳左移16位之后，符号位是1，就代码这个值为负数// 低16位在表示当前正在扩容的线程有多少个,// 为什么低位值为2时，代表有一个线程正在扩容// 每一个线程扩容完毕后，会对低16位进行-1操作，当最后一个线程扩容完毕后，减1的结果还是-1，// 当值为-1时，要对老数组进行一波扫描，查看是否有遗漏的数据没有迁移到新数组else if (pareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))// 调用transfer方法，并且将第二个参数设置为null，就代表是第一次来扩容！transfer(tab, null);}}}

3.5 transfer方法-计算每个线程迁移的长度

// 开始扩容 tab=oldTableprivate final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// n = 数组长度// stride = 每个线程一次性迁移多少数据到新数组int n = tab.length, stride;// 基于CPU的内核数量来计算，每个线程一次性迁移多少长度的数据最合理// NCPU = 4// 举个栗子：数组长度为1024 - 512 - 256 - 128 / 4 = 32// MIN_TRANSFER_STRIDE = 16,为每个线程迁移数据的最小长度if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // 根据CPU计算每个线程一次迁移多长的数据到新数组，如果结果大于16，使用计算结果。 如果结果小于16，就使用最小长度16}

3.6 transfer方法-构建新数组并查看标识属性

// 以32长度数组扩容到64位例子private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// n = 老数组长度 32// stride = 步长 16// 第一个进来扩容的线程需要把新数组构建出来if (nextTab == null) {try {// 将原数组长度左移一位，构建新数组长度Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];// 赋值操作nextTab = nt;} catch (Throwable ex) {// 到这说明已经达到数组长度的最大取值范围sizeCtl = Integer.MAX_VALUE;// 设置sizeCtl后直接结束return;}// 将成员变量的新数组赋值nextTable = nextTab;// 迁移数据时，用到的标识，默认值为老数组长度transferIndex = n; // 32}// 新数组长度int nextn = nextTab.length; // 64// 在老数组迁移完数据后，做的标识ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// 迁移数据时，需要用到的标识boolean advance = true;boolean finishing = false; // 省略部分代码}

3.7 transfer方法-线程领取迁移任务

// 以32长度扩容到64位为例子private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// n：32// stride：16int n = tab.length, stride;if (nextTab == null) {// 省略部分代码…………// nextTable：新数组nextTable = nextTab;// transferIndex：0transferIndex = n;}// nextn：64int nextn = nextTab.length;ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// advance：true，代表当前线程需要接收任务，然后再执行迁移， 如果为false，代表已经接收完任务boolean advance = true;// finishing：false，是否迁移结束！boolean finishing = false; // 循环……// i = 15代表当前线程迁移数据的索引值！！// bound = 0for (int i = 0, bound = 0;;) {// f = null// fh = 0Node<K,V> f; int fh;// 当前线程要接收任务while (advance) {// nextIndex = 16// nextBound = 16int nextIndex, nextBound;// 第一次进来，这两个判断肯定进不去。// 对i进行--，并且判断当前任务是否处理完毕！if (--i >= bound || finishing)advance = false;// 判断transferIndex是否小于等于0，代表没有任务可领取，结束了。// 在线程领取任务会，会对transferIndex进行修改，修改为transferIndex - stride// 在任务都领取完之后，transferIndex肯定是小于等于0的，代表没有迁移数据的任务可以领取else if ((nextIndex = transferIndex) <= 0) {i = -1;advance = false;}// 当前线程尝试领取任务else if (pareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ? nextIndex - stride : 0))) {// 对bound赋值bound = nextBound;// 对i赋值i = nextIndex - 1;// 设置advance设置为false，代表当前线程领取到任务了。advance = false;}}// 开始迁移数据，并且在迁移完毕后，会将advance设置为true}}

3.8 transfer方法-迁移结束操作

// 以32长度扩容到64位为例子private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {for (int i = 0, bound = 0;;) {while (advance) {// 判断扩容是否已经结束！// i < 0：当前线程没有接收到任务！// i >= n: 迁移的索引位置，不可能大于数组的长度，不会成立// i + n >= nextn：因为i最大值就是数组索引的最大值，不会成立if (i < 0 || i >= n || i + n >= nextn) {// 如果进来，代表当前线程没有接收到任务int sc;// finishing为true，代表扩容结束if (finishing) {// 将nextTable新数组设置为nullnextTable = null;// 将当前数组的引用指向了新数组~table = nextTab;// 重新计算扩容阈值 64 - 16 = 48sizeCtl = (n << 1) - (n >>> 1);// 结束扩容return;}// 当前线程没有接收到任务，让当前线程结束扩容操作。// 采用CAS的方式，将sizeCtl - 1，代表当前并发扩容的线程数 - 1if (pareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// sizeCtl的高16位是基于数组长度计算的扩容戳，低16位是当前正在扩容的线程个数if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 代表当前线程并不是最后一个退出扩容的线程，直接结束当前线程扩容return;// 如果是最后一个退出扩容的线程，将finishing和advance设置为truefinishing = advance = true;// 将i设置为老数组长度，让最后一个线程再从尾到头再次检查一下，是否数据全部迁移完毕。i = n; }}// 开始迁移数据，并且在迁移完毕后，会将advance设置为true }}

3.9 transfer方法-迁移数据（链表）

// 以32长度扩容到64位为例子private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {// 省略部分代码…………for (int i = 0, bound = 0;;) {// 省略部分代码…………if (i < 0 || i >= n || i + n >= nextn) {// 省略部分代码…………}// 开始迁移数据，并且在迁移完毕后，会将advance设置为true // 获取指定i位置的Node对象，并且判断是否为nullelse if ((f = tabAt(tab, i)) == null)// 当前桶位置没有数据，无需迁移，直接将当前桶位置设置为fwdadvance = casTabAt(tab, i, null, fwd);// 拿到当前i位置的hash值，如果为MOVED，证明数据已经迁移过了。else if ((fh = f.hash) == MOVED)// 一般是给最后扫描时，使用的判断，如果迁移完毕，直接跳过当前位置。advance = true; // already processedelse {// 当前桶位置有数据，先锁住当前桶位置。synchronized (f) {// 判断之前取出的数据是否为当前的数据。if (tabAt(tab, i) == f) {// ln：null - lowNode// hn：null - highNodeNode<K,V> ln, hn;// hash大于0，代表当前Node属于正常情况，不是红黑树，使用链表方式迁移数据if (fh >= 0) {// lastRun机制// 000000000010000// 这种运算结果只有两种，要么是0，要么是nint runBit = fh & n;// 将f赋值给lastRunNode<K,V> lastRun = f;// 循环的目的就是为了得到链表下经过hash & n结算，结果一致的最后一些数据// 在迁移数据时，值需要迁移到lastRun即可，剩下的指针不需要变换。for (Node<K,V> p = f.next; p != null; p = p.next) {int b = p.hash & n;if (b != runBit) {runBit = b;lastRun = p;}}// runBit == 0，赋值给lnif (runBit == 0) {ln = lastRun;hn = null;}// rubBit == n,赋值给hnelse {hn = lastRun;ln = null;}// 循环到lastRun指向的数据即可，后续不需要再遍历for (Node<K,V> p = f; p != lastRun; p = p.next) {// 获取当前Node的hash值，key值，value值。int ph = p.hash; K pk = p.key; V pv = p.val;// 如果hash&n为0，挂到lowNode上if ((ph & n) == 0)ln = new Node<K,V>(ph, pk, pv, ln);// 如果hash&n为n，挂到highNode上elsehn = new Node<K,V>(ph, pk, pv, hn);}// 采用CAS的方式，将ln挂到新数组的原位置setTabAt(nextTab, i, ln);// 采用CAS的方式，将hn挂到新数组的原位置 + 老数组长度setTabAt(nextTab, i + n, hn);// 采用CAS的方式，将当前桶位置设置为fwdsetTabAt(tab, i, fwd);// advance设置为true，保证可以进入到while循环，对i进行--操作advance = true;}// 省略迁移红黑树的操作}}}}}

3.10 helpTransfer方法-协助扩容

// 在添加数据时，如果插入节点的位置的数据，hash值为-1，代表当前索引位置数据已经被迁移到了新数组// tab：老数组// f：数组上的Node节点final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {// nextTab：新数组// sc：给sizeCtl做临时变量Node<K,V>[] nextTab; int sc;// 第一个判断：老数组不为null// 第二个判断：新数组不为null （将新数组赋值给nextTab）if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {// ConcurrentHashMap正在扩容// 基于老数组长度计算扩容戳int rs = resizeStamp(tab.length);// 第一个判断：fwd中的新数组，和当前正在扩容的新数组是否相等。 相等：可以协助扩容。不相等：要么扩容结束，要么开启了新的扩容// 第二个判断：老数组是否改变了。相等：可以协助扩容。不相等：扩容结束了// 第三个判断：如果正在扩容，sizeCtl肯定为负数，并且给sc赋值while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) {// 第一个判断：将sc右移16位，判断是否与扩容戳一致。 如果不一致，说明扩容长度不一样，退出协助扩容// 第二个、三个判断是BUG：/*sc == rs << 16 + 1 ||如果+1和当前sc一致，说明扩容已经到了最后检查的阶段sc == rs << 16 + MAX_RESIZERS || 判断协助扩容的线程是否已经达到了最大值*/// 第四个判断：transferIndex是从高索引位置到低索引位置领取数据的一个核心属性，如果满足 小于等于0，说明任务被领光了。if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || transferIndex <= 0)// 不需要协助扩容break;// 将sizeCtl + 1，进来协助扩容if (pareAndSwapInt(this, SIZECTL, sc, sc + 1)) {// 协助扩容transfer(tab, nextTab);break;}}return nextTab;}return table;}

四、红黑树操作

在前面搞定了关于数据+链表的添加和扩容操作，现在要搞定红黑树。因为红黑树的操作有点乱，先对红黑树结构有一定了解。

4.1 什么是红黑树

红黑树是一种特殊的平衡二叉树，首选具备了平衡二叉树的特点：左子树和右子数的高度差不会超过1，如果超过了，平衡二叉树就会基于左旋和右旋的操作，实现自平衡。

红黑树在保证自平衡的前提下，还保证了自己的几个特性：

每个节点必须是红色或者黑色。根节点必须是黑色。如果当前节点是红色，子节点必须是黑色所有叶子节点都是黑色。从任意节点到每个叶子节点的路径中，黑色节点的数量是相同的。

当对红黑树进行增删操作时，可能会破坏平衡或者是特性，这是红黑树就需要基于左旋、右旋、变色来保证平衡和特性。

左旋操作：

右旋操作：

变色操作：节点的颜色从黑色变为红色，或者从红色变为黑色，就成为变色。变色操作是在增删数据之后，可能出现的操作。插入数据时，插入节点的颜色一般是红色，因为插入红色节点的破坏红黑树结构的可能性比较低的。如果破坏了红黑树特性，会通过变色来调整

红黑树相对比较复杂，完整的红黑树代码400~500行内容，没有必要全部记下来，或者首先红黑树。

如果向细粒度掌握红黑树的结构：/subject/21?courseNo=339

4.2 TreeifyBin方法-封装TreeNode和双向链表

// 将链表转为红黑树的准备操作private final void treeifyBin(Node<K,V>[] tab, int index) {// b：当前索引位置的NodeNode<K,V> b; int sc;if (tab != null) {// 省略部分代码// 开启链表转红黑树操作// 当前桶内有数据，并且是链表结构else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {// 加锁，保证线程安全synchronized (b) {// 再次判断数据是否有变化，DCLif (tabAt(tab, index) == b) {// 开启准备操作，将之前的链表中的每一个Node，封装为TreeNode，作为双向链表// hd：是整个双向链表的第一个节点。 // tl：是单向链表转换双向链表的临时存储变量TreeNode<K,V> hd = null, tl = null;for (Node<K,V> e = b; e != null; e = e.next) {TreeNode<K,V> p = new TreeNode<K,V>(e.hash, e.key, e.val, null, null);if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}// hd就是整个双向链表// TreeBin的有参构建，将双向链表转为了红黑树。setTabAt(tab, index, new TreeBin<K,V>(hd));}}}}}

4.3 TreeBin有参构造-双向链表转为红黑树

TreeBin中不但保存了红黑树结构，同时还保存在一套双向链表

// 将双向链表转为红黑树的操作。 b：双向链表的第一个节点// TreeBin继承自Node，root：代表树的根节点，first：双向链表的头节点TreeBin(TreeNode<K,V> b) {// 构建Node，并且将hash值设置为-2super(TREEBIN, null, null, null);// 将双向链表的头节点赋值给firstthis.first = b;// 声明r的TreeNode，最后会被赋值为根节点TreeNode<K,V> r = null;// 遍历之前封装好的双向链表for (TreeNode<K,V> x = b, next; x != null; x = next) {next = (TreeNode<K,V>)x.next;// 先将左右子节点清空x.left = x.right = null;// 如果根节点为null，第一次循环if (r == null) {// 将第一个节点设置为当前红黑树的根节点x.parent = null; // 根节点没父节点x.red = false; // 不是红色，是黑色r = x; // 将当前节点设置为r}// 已经有根节点，当前插入的节点要作为父节点的左子树或者右子树else {// 拿到了当前节点key和hash值。K k = x.key;int h = x.hash;Class<?> kc = null;// 循环？for (TreeNode<K,V> p = r;;) {// dir：如果为-1，代表要插入到父节点的左边，如果为1，代表要插入的父节点的右边// ph：是父节点的hash值int dir, ph;// pk：是父节点的keyK pk = p.key;// 父节点的hash值，大于当前节点的hash值，就设置为-1，代表要插入到父节点的左边if ((ph = p.hash) > h)dir = -1;// 父节点的hash值，小于当前节点的hash值，就设置为1，代表要插入到父节点的右边else if (ph < h)dir = 1;// 父节点的hash值和当前节点hash值一致，基于compare方式判断到底放在左子树还是右子树else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||(dir = compareComparables(kc, k, pk)) == 0)dir = tieBreakOrder(k, pk);// 拿到当前父节点。TreeNode<K,V> xp = p;// 将p指向p的left、right，并且判断是否为null// 如果为null，代表可以插入到这位置。if ((p = (dir <= 0) ? p.left : p.right) == null) {// 进来就说明找到要存放当前节点的位置了// 将当前节点的parent指向父节点x.parent = xp;// 根据dir的值，将父节点的left、right指向当前节点if (dir <= 0)xp.left = x;elsexp.right = x;// 插入一个节点后，做一波平衡操作r = balanceInsertion(r, x);break;}}}}// 将根节点复制给rootthis.root = r;// 检查红黑树结构assert checkInvariants(root);}

4.4 balanceInsertion方法-保证红黑树平衡以及特性

// 红黑树的插入动画：https://www.cs.usfca.edu/~galles/visualization/RedBlack.html// 红黑树做自平衡以及保证特性的操作。 root：根节点， x：当前节点static <K,V> TreeNode<K,V> balanceInsertion(TreeNode<K,V> root, TreeNode<K,V> x) {// 先将节点置位红色x.red = true;// xp：父节点// xpp：爷爷节点// xppl：爷爷节点的左子树// xxpr：爷爷节点的右子树for (TreeNode<K,V> xp, xpp, xppl, xppr;;) {// 拿到父节点，并且父节点为红if ((xp = x.parent) == null) {// 当前节点为根节点，置位黑色x.red = false;return x;}// 父节点不是红色，爷爷节点为nullelse if (!xp.red || (xpp = xp.parent) == null)// 什么都不做，直接返回return root;// =====================================// 左子树的操作if (xp == (xppl = xpp.left)) {// 通过变色满足红黑树特性if ((xppr = xpp.right) != null && xppr.red) {// 叔叔节点和父节点变为黑色xppr.red = false;xp.red = false;// 爷爷节点置位红色xpp.red = true;// 让爷爷节点作为当前节点，再走一次循环x = xpp;}else {// 如果当前节点是右子树，通过父节点的左旋，变为左子树的结构if (x == xp.right) {、// 父节点做左旋操作root = rotateLeft(root, x = xp);xpp = (xp = x.parent) == null ? null : xp.parent;}if (xp != null) {// 父节点变为黑色xp.red = false;if (xpp != null) {// 爷爷节点变为红色xpp.red = true;// 爷爷节点做右旋操作root = rotateRight(root, xpp);}}}}// 右子树（只讲左子树就足够了，因为业务都是一样的）else {if (xppl != null && xppl.red) {xppl.red = false;xp.red = false;xpp.red = true;x = xpp;}else {if (x == xp.left) {root = rotateRight(root, x = xp);xpp = (xp = x.parent) == null ? null : xp.parent;}if (xp != null) {xp.red = false;if (xpp != null) {xpp.red = true;root = rotateLeft(root, xpp);}}}}}}

4.5 putTreeVal方法-添加节点

整体操作就是判断当前节点要插入到左子树，还是右子数，还是覆盖操作。

确定左子树和右子数之后，直接维护双向链表和红黑树结构，并且再判断是否需要自平衡。

TreeBin的双向链表用的头插法。

// 添加节点到红黑树内部final TreeNode<K,V> putTreeVal(int h, K k, V v) {// Class对象Class<?> kc = null;// 搜索节点boolean searched = false;// 死循环，p节点是根节点的临时引用for (TreeNode<K,V> p = root;;) {// dir：确定节点是插入到左子树还是右子数// ph：父节点的hash值// pk：父节点的keyint dir, ph; K pk;// 根节点是否为诶null，把当前节点置位根节点if (p == null) {first = root = new TreeNode<K,V>(h, k, v, null, null);break;}// 判断当前节点要放在左子树还是右子数else if ((ph = p.hash) > h)dir = -1;else if (ph < h)dir = 1;// 如果key一致，直接返回p，由putVal去修改数据else if ((pk = p.key) == k || (pk != null && k.equals(pk)))return p;// hash值一致，但是key的==和equals都不一样，基于Compare去判断else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||// 基于compare判断也是一致，就进到if判断(dir = compareComparables(kc, k, pk)) == 0) {// 开启搜索，查看是否有相同的key，只有第一次循环会执行。if (!searched) {TreeNode<K,V> q, ch;searched = true;if (((ch = p.left) != null &&(q = ch.findTreeNode(h, k, kc)) != null) ||((ch = p.right) != null &&(q = ch.findTreeNode(h, k, kc)) != null))// 如果找到直接返回return q;}// 再次判断hash大小，如果小于等于，返回-1dir = tieBreakOrder(k, pk);}// xp是父节点的临时引用TreeNode<K,V> xp = p;// 基于dir判断是插入左子树还有右子数，并且给p重新赋值if ((p = (dir <= 0) ? p.left : p.right) == null) {// first引用拿到TreeNode<K,V> x, f = first;// 将当前节点构建出来first = x = new TreeNode<K,V>(h, k, v, f, xp);// 因为当前的TreeBin除了红黑树还维护这一个双向链表，维护双向链表的操作if (f != null)f.prev = x;// 维护红黑树操作if (dir <= 0)xp.left = x;elsexp.right = x;// 如果如节点是黑色的，当前节点红色即可，说明现在插入的节点没有影响红黑树的平衡if (!xp.red)x.red = true;else {// 说明插入的节点是黑色的// 加锁操作lockRoot();try {// 自平衡一波。root = balanceInsertion(root, x);} finally {// 释放锁操作unlockRoot();}}break;}}// 检查一波红黑树结构assert checkInvariants(root);// 代表插入了新节点return null;}

4.6 TreeBin的锁操作

TreeBin的锁操作，没有基于AQS，仅仅是对一个变量的CAS操作和一些业务判断实现的。

每次读线程操作，对lockState+4。

写线程操作，对lockState + 1，如果读操作占用着线程，就先+2，waiter是当前线程，并挂起当前线程

// TreeBin的锁操作// 如果说有读线程在读取红黑树的数据，这时，写线程要阻塞（做平衡前）// 如果有写线程正在操作红黑树（做平衡），读线程不会阻塞，会读取双向链表// 读读不会阻塞！static final class TreeBin<K,V> extends Node<K,V> {// waiter：等待获取写锁的线程volatile Thread waiter;// lockState：当前TreeBin的锁状态volatile int lockState;// 对锁状态进行运算的值// 有线程拿着写锁static final int WRITER = 1; // 有写线程，再等待获取写锁static final int WAITER = 2; // 读线程，在红黑树中检索时，需要先对lockState + READER// 这个只会在读操作中遇到static final int READER = 4; // 加锁-写锁private final void lockRoot() {// 将lockState从0设置为1，代表拿到写锁成功if (!pareAndSwapInt(this, LOCKSTATE, 0, WRITER))// 如果写锁没拿到，执行contendedLockcontendedLock(); }// 释放写锁private final void unlockRoot() {lockState = 0;}// 写线程没有拿到写锁，执行当前方法private final void contendedLock() {// 是否有线程正在等待boolean waiting = false;// 死循环，s是lockState的临时变量for (int s;;) {// // lockState & 11111101 ,只要结果为0，说明当前写锁，和读锁都没线程获取if (((s = lockState) & ~WAITER) == 0) {// CAS一波，尝试将lockState再次修改为1，if (pareAndSwapInt(this, LOCKSTATE, s, WRITER)) {// 成功拿到锁资源，并判断是否在waitingif (waiting)// 如果当前线程挂起过，直接将之前等待的线程资源设置为nullwaiter = null;return;}}// 有读操作在占用资源// lockState & 00000010,代表当前没有写操作挂起等待。else if ((s & WAITER) == 0) {// 基于CAS，将LOCKSTATE的第二位设置为1if (pareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {// 如果成功，代表当前线程可以waiting等待了waiting = true;waiter = Thread.currentThread();}}else if (waiting)// 挂起当前线程！会由写操作唤醒LockSupport.park(this);}}}

4.7 transfer迁移数据

首先红黑结构的数据迁移是基于双向链表封装的数据。

如果高低位的长度小于等于6，封装为链表迁移到新数组

如果高低位的长度大于6，依然封装为红黑树迁移到新数组

// 红黑树的迁移操作单独拿出来，TreeBin中不但有红黑树，还有双向链表，迁移的过程是基于双向链表迁移TreeBin<K,V> t = (TreeBin<K,V>)f;// lo，hi扩容后要放到新数组的高低位的链表TreeNode<K,V> lo = null, loTail = null;TreeNode<K,V> hi = null, hiTail = null;// lc，hc在记录高低位数据的长度int lc = 0, hc = 0;// 遍历TreeBin中的双向链表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;elseloTail.next = p;loTail = p;// 低位长度++++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.next = p;hiTail = p;// 高位长度++++hc;}}// 封装低位节点，如果低位节点的长度小于等于6，转回成链表。 如果长度大于6，需要重新封装红黑树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);// 当前位置数据迁移完毕，设置上fwdsetTabAt(tab, i, fwd);// 开启前一个节点的数据迁移advance = true;

五、查询数据

5.1 get方法-查询数据的入口

在查询数据时，会先判断当前key对应的value，是否在数组上。

其次会判断当前位置是否属于特殊情况：数据被迁移、位置被占用、红黑树结构

最后判断链表上是否有对应的数据。

找到返回指定的value，找不到返回null即可

// 基于key查询valuepublic V get(Object key) {// tab：数组， e：查询指定位置的节点 n：数组长度Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;// 基于传入的key，计算hash值int h = spread(key.hashCode());// 数组不为null，数组上得有数据，拿到指定位置的数组上的数据if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {// 数组上数据恩地hash值，是否和查询条件key的hash一样if ((eh = e.hash) == h) {// key的==或者equals是否一致，如果一致，数组上就是要查询的数据if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}// 如果数组上的数据的hash为负数，有特殊情况，else if (eh < 0)// 三种情况，数据迁移走了，节点位置被占，红黑树return (p = e.find(h, key)) != null ? p.val : null;// 肯定走链表操作while ((e = e.next) != null) {// 如果hash值一致，并且key的==或者equals一致，返回当前链表位置的数据if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}// 如果上述三个流程都没有知道指定key对应的value，那就是key不存在，返回null即可return null;}

5.2 ForwardingNode的find方法

在查询数据时，如果发现已经扩容了，去新数组上查询数据

在数组和链表上正常找key对应的value

可能依然存在特殊情况：

再次是fwd，说明当前线程可能没有获取到CPU时间片，导致CHM再次触发扩容，重新走当前方法可能是被占用或者是红黑树，再次走另外两种find方法的逻辑

// 在查询数据时，发现当前桶位置已经放置了fwd，代表已经被迁移到了新数组Node<K,V> find(int h, Object k) {// key：get(key) h：key的hash tab：新数组outer: for (Node<K,V>[] tab = nextTable;;) {// n：新数组长度， e：新数组上定位的位置上的数组Node<K,V> e; int n;if (k == null || tab == null || (n = tab.length) == 0 || (e = tabAt(tab, (n - 1) & h)) == null)return null;// 开始在新数组中走逻辑for (;;) {// eh：新数组位置的数据的hashint eh; K ek;// 判断hash是否一致，如果一致，再判断==或者equals。if ((eh = e.hash) == h && ((ek = e.key) == k || (ek != null && k.equals(ek))))// 在新数组找到了数据return e;// 发现到了新数组，hash值又小于0if (eh < 0) {// 套娃，发现刚刚在扩容，到了新数组，发现又扩容if (e instanceof ForwardingNode) {// 再次重新走最外层循环，拿到最新的nextTabletab = ((ForwardingNode<K,V>)e).nextTable;continue outer;}else// 占了，红黑树return e.find(h, k);}// 说明不在数组上，往下走链表if ((e = e.next) == null)// 进来说明链表没找到，返回nullreturn null;}}}

5.3 ReservationNode的find方法

没什么说的，直接返回null

因为当前桶位置被占用的话，说明数据还没放到当前位置，当前位置可以理解为就是null

Node<K,V> find(int h, Object k) {return null;}

5.4 TreeBin的find方法

在红黑树中执行find方法后，会有两个情况

如果有线程在持有写锁或者等待获取写锁，当前查询就要在双向链表中锁检索如果没有线程持有写锁或者等待获取写锁，完全可以对lockState + 4，然后去红黑树中检索，并且在检索完毕后，需要对lockState - 4，再判断是否需要唤醒等待写锁的线程

// 在红黑树中检索数据final Node<K,V> find(int h, Object k) {// 非空判断if (k != null) {// e：Treebin中的双向链表，for (Node<K,V> e = first; e != null; ) {int s; K ek;// s：TreeBin的锁状态// 00000010// 00000001if (((s = lockState) & (WAITER|WRITER)) != 0) {// 如果进来if，说明要么有写线程在等待获取写锁，要么是由写线程持有者写锁// 如果出现这个情况，他会去双向链表查询数据if (e.hash == h && ((ek = e.key) == k || (ek != null && k.equals(ek))))return e;e = e.next;}// 说明没有线程等待写锁或者持有写锁，将lockState + 4，代表当前读线程可以去红黑树中检索数据else if (pareAndSwapInt(this, LOCKSTATE, s, s + READER)) {TreeNode<K,V> r, p;try {// 基于findTreeNode在红黑树中检索数据p = ((r = root) == null ? null : r.findTreeNode(h, k, null));} finally {Thread w;// 会对lockState - 4，读线程拿到数据了，释放读锁// 可以确认，如果-完4，等于WAITER，说明有写线程可能在等待，判断waiter是否为nullif (U.getAndAddInt(this, LOCKSTATE, -READER) == (READER|WAITER) && (w = waiter) != null)// 当前我是最后一个在红黑树中检索的线程，同时有线程在等待持有写锁，唤醒等待的写线程LockSupport.unpark(w);}return p;}}}return null;}

5.6 TreeNode的findTreeNode方法

红黑树的检索方式，套路很简单，及时基于hash值，来决定去找左子树还有右子数。

如果hash值一致，判断是否 == 、equals，满足就说明找到数据

如果hash值一致，并不是找的数据，基于compare方式，再次决定找左子树还是右子数，知道找到当前节点的子节点为null，停住。

// 红黑树中的检索方法final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {if (k != null) {TreeNode<K,V> p = this;do {int ph, dir; K pk; TreeNode<K,V> q;// 声明左子树和右子数TreeNode<K,V> pl = p.left, pr = p.right;// 直接比较hash值，来决决定走左子树还是右子数if ((ph = p.hash) > h)p = pl;else if (ph < h)p = pr;// 判断当前的子树是否和查询的k == 或者equals，直接返回else if ((pk = p.key) == k || (pk != null && k.equals(pk)))return p;else if (pl == null)p = pr;else if (pr == null)p = pl;else if ((kc != null ||(kc = comparableClassFor(k)) != null) &&(dir = compareComparables(kc, k, pk)) != 0)p = (dir < 0) ? pl : pr;// 递归继续往底层找else if ((q = pr.findTreeNode(h, k, kc)) != null)return q;elsep = pl;} while (p != null);}return null;}

六、ConcurrentHashMap其他方法

6.1 compute方法

修改ConcurrentHashMap中指定key的value时，一般会选择先get出来，然后再拿到原value值，基于原value值做一些修改，最后再存放到咱们ConcurrentHashMap

public static void main(String[] args) {ConcurrentHashMap<String,Integer> map = new ConcurrentHashMap();map.put("key",1);// 修改key对应的value，追加上1// 之前的操作方式Integer oldValue = (Integer) map.get("key");Integer newValue = oldValue + 1;map.put("key",newValue);System.out.println(map);// 现在的操作方式pute("key",(key,computeOldValue) -> {if(computeOldValue == null){computeOldValue = 0;}return computeOldValue + 1;});System.out.println(map);}

6.2 compute方法源码分析

整个流程和putVal方法很类似，但是内部涉及到了占位的情况RESERVED

整个compute方法和putVal的区别就是，compute方法的value需要计算，如果key存在，基于oldValue计算出新结果，如果key不存在，直接基于oldValue为null的情况，去计算新的value。

// compute 方法public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {if (key == null || remappingFunction == null)throw new NullPointerException();// 计算key的hashint h = spread(key.hashCode());V val = null;int delta = 0;int binCount = 0;// 初始化，桶上赋值，链表插入值，红黑树插入值for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;// 初始化if (tab == null || (n = tab.length) == 0)tab = initTable();// 桶上赋值else if ((f = tabAt(tab, i = (n - 1) & h)) == null) {// 数组指定的索引位置是没有数据，当前数据必然要放到数组上。// 因为value需要计算得到，计算的时间不可估计，所以这里并没有通过CAS的方式处理并发操作，直接添加临时占用节点，// 并占用当前临时节点的锁资源。Node<K,V> r = new ReservationNode<K,V>();synchronized (r) {// 以CAS的方式将数据放上去if (casTabAt(tab, i, null, r)) {binCount = 1;Node<K,V> node = null;try {// 如果ReservationNode临时Node存放成功，直接开始计算valueif ((val = remappingFunction.apply(key, null)) != null) {delta = 1;// 将计算的value和传入的key封装成一个新Node，通过CAS存储到当前数组上node = new Node<K,V>(h, key, val, null);}} finally {setTabAt(tab, i, node);}}}if (binCount != 0)break;}else {// 省略部分代码。主要是针对在链表上的替换、添加，以及在红黑树上的替换、添加}}if (delta != 0)addCount((long)delta, binCount);return val;}

6.3 computeIfPresent、computeIfAbsent、compute区别

compute的BUG，如果在计算结果的函数中，又涉及到了当前的key，会造成死锁问题。

public static void main(String[] args) {ConcurrentHashMap<String,Integer> map = new ConcurrentHashMap();pute("key",(k,v) -> {return pute("key",(key,value) -> {return 1111;});});System.out.println(map);}

computeIfPresent和computeIfAbsent其实就是将compute方法拆开成了两个方法

compute会在key不存在时，正常存放结果，如果key存在，就基于oldValue计算newValue

computeIfPresent：要求key在map中必须存在，需要基于oldValue计算newValue

computeIfAbsent：要求key在map中不能存在，必须为null，才会基于函数得到value存储进去

computeIfPresent：

// 如果key存在，才执行修改操作public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;if (tab == null || (n = tab.length) == 0)tab = initTable();// 如果key不存在，什么事都不做~else if ((f = tabAt(tab, i = (n - 1) & h)) == null)break;else {synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) {binCount = 1;for (Node<K,V> e = f, pred = null;; ++binCount) {K ek;// 如果查看到有 == 或者equals的key，就直接修改即可if (e.hash == h &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {val = remappingFunction.apply(key, e.val);if (val != null)e.val = val;else {delta = -1;Node<K,V> en = e.next;if (pred != null)pred.next = en;elsesetTabAt(tab, i, en);}break;}pred = e;// 走完链表，还是没找到指定数据，直接break;if ((e = e.next) == null)break;}}// 省略部分代码return val;}

computeIfAbsent核心位置源码：

// key必须不存在才会执行添加操作public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {for (Node<K,V>[] tab = table;;) {else if ((f = tabAt(tab, i = (n - 1) & h)) == null) {// 如果key不存在，正常添加;Node<K,V> r = new ReservationNode<K,V>();synchronized (r) {if (casTabAt(tab, i, null, r)) {binCount = 1;Node<K,V> node = null;try {if ((val = mappingFunction.apply(key)) != null)node = new Node<K,V>(h, key, val, null);} finally {setTabAt(tab, i, node);}}}}else {boolean added = false;synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) {binCount = 1;for (Node<K,V> e = f;; ++binCount) {K ek; V ev;// 如果key存在，直接break;if (e.hash == h &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {val = e.val;break;}// 如果没有找到一样的key，计算value结果接口Node<K,V> pred = e;if ((e = e.next) == null) {if ((val = mappingFunction.apply(key)) != null) {added = true;pred.next = new Node<K,V>(h, key, val, null);}break;}}}// 省略部分代码 return val;}

6.4 replace方法详解

涉及到类似CAS的操作，需要将ConcurrentHashMap的value从val1改为val2的场景就可以使用replace实现。

replace内部要求key必须存在，替换value值之前，要先比较oldValue，只有oldValue一致时，才会完成替换操作。

// replace方法调用的replaceNode方法， value：newValue， cv：oldValuefinal V replaceNode(Object key, V value, Object cv) {int hash = spread(key.hashCode());for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;// 在数组没有初始化时，或者key不存在时，什么都不干。if (tab == null || (n = tab.length) == 0 ||(f = tabAt(tab, i = (n - 1) & hash)) == null)break;else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {V oldVal = null;boolean validated = false;synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) {validated = true;for (Node<K,V> e = f, pred = null;;) {K ek;// 找到key一致的Node了。if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {// 拿到当前节点的原值。V ev = e.val;// 拿oldValue和原值做比较，如果一致，if (cv == null || cv == ev || (ev != null && cv.equals(ev))) {// 可以开始替换oldVal = ev;if (value != null)e.val = value;else if (pred != null)pred.next = e.next;elsesetTabAt(tab, i, e.next);}break;}pred = e;if ((e = e.next) == null)break;}}else if (f instanceof TreeBin) {validated = true;TreeBin<K,V> t = (TreeBin<K,V>)f;TreeNode<K,V> r, p;if ((r = t.root) != null &&(p = r.findTreeNode(hash, key, null)) != null) {V pv = p.val;if (cv == null || cv == pv ||(pv != null && cv.equals(pv))) {oldVal = pv;if (value != null)p.val = value;else if (t.removeTreeNode(p))setTabAt(tab, i, untreeify(t.first));}}}}}if (validated) {if (oldVal != null) {if (value == null)addCount(-1L, -1);return oldVal;}break;}}}return null;}

6.5 merge方法详解

merge(key,value,Function<oldValue,value>);

在使用merge时，有三种情况可能发生：

如果key不存在，就跟put(key,value);如果key存在，就可以基于Function计算，得到最终结果 结果不为null，将key对应的value，替换为Function的结果结果为null，删除当前key

分析merge源码

public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {if (key == null || value == null || remappingFunction == null) throw new NullPointerException();int h = spread(key.hashCode());V val = null;int delta = 0;int binCount = 0;for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;if (tab == null || (n = tab.length) == 0)tab = initTable();// key不存在，直接执行正常的添加操作，将value作为值，添加到hashMapelse if ((f = tabAt(tab, i = (n - 1) & h)) == null) {if (casTabAt(tab, i, null, new Node<K,V>(h, key, value, null))) {delta = 1;val = value;break;}}else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {synchronized (f) {if (tabAt(tab, i) == f) {if (fh >= 0) {binCount = 1;for (Node<K,V> e = f, pred = null;; ++binCount) {K ek;// 判断链表中，有当前的keyif (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {// 基于函数，计算valueval = remappingFunction.apply(e.val, value);// 如果计算的value不为null，正常替换if (val != null)e.val = val;// 计算的value是null，直接让上一个指针指向我的next，绕过当前节点else {delta = -1;Node<K,V> en = e.next;if (pred != null)pred.next = en;elsesetTabAt(tab, i, en);}break;}pred = e;if ((e = e.next) == null) {delta = 1;val = value;pred.next =new Node<K,V>(h, key, val, null);break;}}}else if (f instanceof TreeBin) {binCount = 2;TreeBin<K,V> t = (TreeBin<K,V>)f;TreeNode<K,V> r = t.root;TreeNode<K,V> p = (r == null) ? null :r.findTreeNode(h, key, null);val = (p == null) ? value :remappingFunction.apply(p.val, value);if (val != null) {if (p != null)p.val = val;else {delta = 1;t.putTreeVal(h, key, val);}}else if (p != null) {delta = -1;if (t.removeTreeNode(p))setTabAt(tab, i, untreeify(t.first));}}}}if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);break;}}}if (delta != 0)addCount((long)delta, binCount);return val;}

七、ConcurrentHashMap计数器

一、addCount方法分析

addCount方法本身就是为了记录ConcurrentHashMap中元素的个数。

两个方向组成：

计数器，如果添加元素成功，对计数器 + 1检验当前ConcurrentHashMap是否需要扩容

计数器选择的不是AtomicLong，而是类似LongAdder的一个功能

addCount源码分析

private final void addCount(long x, int check) {// ================================计数=====================================// as： CounterCell[]// s：是自增后的元素个数// b：原来的baseCountCounterCell[] as; long b, s;// 判断CounterCell不为null，代表之前有冲突问题，有冲突直接进到if中// 如果CounterCell[]为null，直接执行||后面的CAS操作，直接修改baseCountif ((as = counterCells) != null ||// 如果对baseCount++成功。直接告辞。 如果CAS失败，直接进到if中!pareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {// 导致，说明有并发问题。// 进来的方式有两种：// 1. counterCell[] 有值。// 2. counterCell[] 无值，但是CAS失败。// m：数组长度 - 1// a：当前线程基于随机数，获得到的数组上的某一个CounterCellCounterCell a; long v; int m;// 是否有冲突，默认为true，代表没有冲突boolean uncontended = true;// 判断CounterCell[]没有初始化，执行fullAddCount方法，初始化数组if (as == null || (m = as.length - 1) < 0 ||// CounterCell[]已经初始化了，基于随机数拿到数组上的一个CounterCell，如果为null，执行fullAddCount方法，初始化CounterCell(a = as[ThreadLocalRandom.getProbe() & m]) == null ||// CounterCell[]已经初始化了，并且指定索引位置上有CounterCell// 直接CAS修改指定的CounterCell上的value即可。// CAS成功，直接告辞！// CAS失败，代表有冲突，uncontended = false，执行fullAddCount方法!(uncontended = pareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {fullAddCount(x, uncontended);return;}// 如果链表长度小于等于1，不去判断扩容if (check <= 1)return;// 将所有CounterCell中记录的信累加，得到最终的元素个数s = sumCount();}// ================================判断扩容=======================================// 判断check大于等于，remove的操作就是小于0的。 因为添加时，才需要去判断是否需要扩容if (check >= 0) {// 一堆小变量Node<K,V>[] tab, nt; int n, sc;// 当前元素个数是否大于扩容阈值，并且数组不为null，数组长度没有达到最大值。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 (pareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}// 没有线程执行扩容，我来扩容else if (pareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);// 重新计数。s = sumCount();}}}// CounterCell的类，就类似于LongAdder的Cell@sun.misc.Contended static final class CounterCell {// volatile修饰的value，并且外部基于CAS的方式修改volatile long value;CounterCell(long x) {value = x; }}@sun.misc.Contended（JDK1.8）：这个注解是为了解决伪共享的问题（解决缓存行同步带来的性能问题）。CPU在操作主内存变量前，会将主内存数据缓存到CPU缓存（L1,L2,L3）中，CPU缓存L1，是以缓存行为单位存储数据的，一般默认的大小为64字节。缓存行同步操作，影响CPU一定的性能。@Contented注解，会将当前类中的属性，会独占一个缓存行，从而避免缓存行失效造成的性能问题。@Contented注解，就是将一个缓存行的后面7个位置，填充上7个没有意义的数据。long value; long l1,l2,l3,l4,l5,l6,l7;// 整体CounterCell数组数据到baseCountfinal long sumCount() {// 拿到CounterCell[]CounterCell[] as = counterCells; CounterCell a;// 拿到baseCountlong sum = baseCount;// 循环走你，遍历CounterCell[]，将值累加到sum中，最终返回sumif (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;}}return sum;}// CounterCell数组没有初始化// CounterCell对象没有构建// 什么都有，但是有并发问题，导致CAS失败private final void fullAddCount(long x, boolean wasUncontended) {// h：当前线程的随机数int h;// 判断当前线程的Probe是否初始化。if ((h = ThreadLocalRandom.getProbe()) == 0) {// 初始化一波ThreadLocalRandom.localInit(); // 生成随机数。h = ThreadLocalRandom.getProbe();// 标记，没有冲突wasUncontended = true;}// 阿巴阿巴boolean collide = false; // 死循环………… for (;;) {// as：CounterCell[]// a：CounterCell对 null// n：数组长度// v：value值CounterCell[] as; CounterCell a; int n; long v;// CounterCell[]不为null时，做CAS操作if ((as = counterCells) != null && (n = as.length) > 0) {// 拿到当前线程随机数对应的CounterCell对象，为null// 第一个if：当前数组已经初始化，但是指定索引位置没有CounterCell对象，构建CounterCell对象放到数组上if ((a = as[h & (n - 1)]) == null) {// 判断cellsBusy是否为0，if (cellsBusy == 0) {// 构建CounterCell对象CounterCell r = new CounterCell(x); // 在此判断cellsBusy为0，CAS从0修改为1，代表可以操作当前数组上的指定索引，构建CounterCell，赋值进去if (cellsBusy == 0 &&pareAndSwapInt(this, CELLSBUSY, 0, 1)) {// 构建未完成boolean created = false;try {// 阿巴阿巴 CounterCell[] rs; int m, j;// DCL，还包含复制if ((rs = counterCells) != null && (m = rs.length) > 0 &&// 再次拿到指定索引位置的值，如果为null，正常将前面构建的CounterCell对象，赋值给数组rs[j = (m - 1) & h] == null) {// 将CounterCell对象赋值到数组rs[j] = r;// 构建完成created = true;}} finally {// 归位cellsBusy = 0;}if (created)// 跳出循环，告辞break;continue; // Slot is now non-empty}}collide = false;}// 指定索引位置上有CounterCell对象，有冲突，修改冲突标识else if (!wasUncontended) wasUncontended = true;// CAS，将数组上存在的CounterCell对象的value进行 + 1操作else if (pareAndSwapLong(a, CELLVALUE, v = a.value, v + x))// 成功，告辞。break;// 之前拿到的数组引用和成员变量的引用值不一样了，// CounterCell数组的长度是都大于CPU内核数，不让CounterCell数组长度大于CPU内核数。else if (counterCells != as || n >= NCPU)// 当前线程的循环失败，不进行扩容collide = false; // 如果没并发问题，并且可以扩容，设置标示位，下次扩容 else if (!collide)collide = true;// 扩容操作// 先判断cellsBusy为0，再基于CAS将cellsBusy从0修改为1。else if (cellsBusy == 0 &&pareAndSwapInt(this, CELLSBUSY, 0, 1)) {try {// DCL!if (counterCells == as) {// 构建一个原来长度2倍的数组CounterCell[] rs = new CounterCell[n << 1];// 将老数组数据迁移到新数组for (int i = 0; i < n; ++i)rs[i] = as[i];// 新数组复制给成员变量counterCells = rs;}} finally {// 归位cellsBusy = 0;}// 归位collide = false;// 开启下次循环continue; }// 重新设置当前线程的随机数，争取下次循环成功！h = ThreadLocalRandom.advanceProbe(h);}// CounterCell[]没有初始化// 判断cellsBusy为0.代表没有其他线程在初始化或者扩容当前CounterCell[]// 判断counterCells还是之前赋值的as，代表没有并发问题else if (cellsBusy == 0 && counterCells == as &&// 修改cellsBusy，从0改为1，代表当前线程要开始初始化了pareAndSwapInt(this, CELLSBUSY, 0, 1)) {// 标识，init未成功boolean init = false;try {// DCL! if (counterCells == as) {// 构建CounterCell[]，默认长度为2CounterCell[] rs = new CounterCell[2];// 用当前线程的随机数，和数组长度 - 1，进行&运算，将这个位置上构建一个CounterCell对象，赋值value为1rs[h & 1] = new CounterCell(x);// 将声明好的rs，赋值给成员变量counterCells = rs;// init成功init = true;}} finally {// cellsBusy归位。cellsBusy = 0;}if (init)// 退出循环break;}// 到这就直接在此操作baseCount。else if (pareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))break;// Fall back on using base}}

二、size方法方法分析

size获取ConcurrentHashMap中的元素个数

public int size() {// 基于sumCount方法获取元素个数long n = sumCount();// 做了一些简单的健壮性判断return ((n < 0L) ? 0 :(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :(int)n);}// 整体CounterCell数组数据到baseCountfinal long sumCount() {// 拿到CounterCell[]CounterCell[] as = counterCells; CounterCell a;// 拿到baseCountlong sum = baseCount;// 循环走你，遍历CounterCell[]，将值累加到sum中，最终返回sumif (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;}}return sum;}