文章目录
- 第一章、线程基础知识
- 一、基础概念
- 1、什么是程序?
- 2、什么是进程?
- 3、什么是线程?
- 4、什么是线程的切换(Context Switch)?
- 5、单核CPU 设定多线程是否有意义?
- 6、工作线程数是不是设置的越大越好?
- 7、工作线程数(线程池中线程数量)设多少合适?
- 二、创建线程的3 种方法
- 1、继承Thread类 重写run方法
- 2、实现Runnable接口 重写run方法
- 3、实现Callable 重写call方法,配合FutureTask
- 4、基于线程池构建线程
- 三、线程状态
- 四、线程的常用方法
- 1、获取当前线程
- 2、线程的名字
- 3、线程的优先级
- 4、线程的让步
- 5、线程的休眠
- 6、线程的强占
- 7、守护线程
- 8、线程的等待和唤醒
- (1)wait notify
- (2)LockSupport
- 五、线程打断interrupt
- 六、线程结束
- stop方法(不用)
- suspend 暂停线程和 resume 恢复线程(不用)
- 使用共享变量(很少会用)
- interrupt方式
- 第二章、并发编程的三大特性
- 一、可见性(visibility)
- 1、什么是可见性
- 2、解决可见性的方式
- (1)volatile
- (2)synchronized
- (3)Lock
- (4)final
- 二、有序性(ordering)
- 1、什么是有序性
- 2、as-if-serial
- 3、happens-before
- 4、volatile
- 三、原子性(atomicity)
- 1、什么是并发编程的原子性
- 2、保证并发编程的原子性
- (1)synchronized
- (2)CAS
- (3)Lock锁
- (4)ThreadLocal
- 第三章、锁
- 一、锁的分类
- 1、可重入锁、不可重入锁
- 2、乐观锁、悲观锁
- 3、公平锁、非公平锁
- 4、互斥锁、共享锁
- 二、深入synchronized
- 1、类锁、对象锁
- 2、synchronized 的优化
- 3、synchronized 实现原理
- 4、synchronized 的锁升级
- 5、重量锁底层ObjectMonitor
- 三、深入ReentrantLock
- 1、ReentrantLock和synchronized的区别
- 2、AQS概述
- 3、加锁流程源码剖析
- (1)加锁流程概述
- (2)三种加锁源码分析
- 4、释放锁流程源码剖析
- (1)释放锁流程概述
- (2)释放锁源码分析
- 5、AQS常见的问题
- (1)AQS中为什么要有一个虚拟的head节点
- (2)AQS中为什么选择使用双向链表,而不是单向链表
- 6、ConditionObject
- (1)ConditionObject的介绍&应用
- (2)Condition的构建方式&核心属性
- (3)Condition的await方法分析(前置分析)
- (4)Condition的signal方法分析
- (5)Conditiond的await方法分析(后置分析)
- (6)Condition的awaitNanos&signalAll方法分析
- 四、深入ReentrantReadWriteLock
- 1、为什么要出现读写锁
- 2、读写锁的实现原理
- 3、写锁分析
- (1)写锁加锁流程概述
- (2)写锁加锁源码分析
- (3)写锁释放锁流程概述&释放锁源码
- 4、读锁分析
- (1)读锁加锁流程概述
- (2)基础读锁流程
- (3)读锁重入流程
- (4)读锁加锁的后续逻辑fullTryAcquireShared
- (5)读线程在AQS队列获取锁资源的后续操作
- (6)读锁的释放锁流程
- 第四章、JUC并发工具
- 一、CountDownLatch应用&源码分析
- 1、CountDownLatch介绍
- 2、CountDownLatch应用
- 3、CountDownLatch源码分析
- (1)有参构造
- (2)await方法
- (3)countDown方法
- 二、CyclicBarrier应用&源码分析
- 1、CyclicBarrier介绍
- 2、CyclicBarrier应用
- 3、CyclicBarrier源码分析
- (1)CyclicBarrier的核心属性
- (2)CyclicBarrier的有参构造
- (3)CyclicBarrier中的await方法
- 三、Semaphone应用&源码分析
- 1、Semaphore介绍
- 2、Semaphore应用
- 3、Semaphore源码分析
- (1)Semaphore的整体结构
- (2)Semaphore的非公平的获取资源
- (3)Semaphore公平实现
- (4)Semaphore释放资源
- 4、AQS 中PROPAGATE 节点
- (1)掌握JDK1.5-Semaphore执行流程图
- (2)分析JDK1.8的变化
第一章、线程基础知识
一、基础概念
1、什么是程序?
存在硬盘里的可执行文件。
2、什么是进程?
当程序得到运行的指令后,把相关信息加载到内存中运行,每一份内存中的运行的程序称之为进程。一个程序可以有多个进程执行,当然也可以代码控制只执行一个进程。
进程是操作系统资源分配的基本单位。
3、什么是线程?
通俗来说,线程就是程序的不同的执行路径。如果程序里没有同时运行的路径,则为单线程,也是主线程(main方法开启的线程)。
当启动了一个程序后,中间会产生分支,不同的分支在同时运行,则为多线程。
线程是进程内部调度执行的基本单位,一个进程可以包含多个线程。进程是内存里静态的概念,线程的动态的概念,真正执行的是线程。多个线程共享进程里需要的资源。解决多线程问题基本就是解决共享资源时产生的问题。
4、什么是线程的切换(Context Switch)?
当操作系统调度CPU 执行一个线程时,线程里的指令放在CPU 里的PC 里,数据放在CPU里的寄存器组(Registers)里,然后CPU 计算单元(ALU)进行计算,输出等操作。然后该操作系统中该线程的时间片结束了之后,会把该线程的指令和数据存到缓存里,然后把另一个线程的指令和数据在存到CPU 进行计算,如此反复。中间需要操作系统的调度过程,也需要消耗资源。
5、单核CPU 设定多线程是否有意义?
当然有意义,线程的不是所有操作都需要CPU 的,如等待IO 等。在线程等待时,是不消耗CPU 的,这是可执行其他线程来充分利用CPU。
线程按执行可分为,CPU 密集型和IO 密集型。
6、工作线程数是不是设置的越大越好?
当然不是,线程之间的切换也是消耗资源的。
7、工作线程数(线程池中线程数量)设多少合适?
Nthreads = Ncpu * Ncpu * (1+W/C)
其中:
Ncpu 是处理器的核的数量,可以通过Runtime.getRuntime().availableProcessors() 得到
Ucpu 是期望的CPU 利用率(该值应介于0 和1 之间)
W/C 是等待时间与计算时间的比率。
具体W/C是多少呢,需要通过性能分析工具(如Profiler)统计出结果。如果运行环境是远程服务端,则用Arthas。
二、创建线程的3 种方法
1、继承Thread类 重写run方法
启动线程是调用start方法,这样会创建一个新的线程,并执行线程的任务。
如果直接调用run方法,这样会让当前线程执行run方法中的业务逻辑。
public class MiTest {
public static void main(String[] args) {
MyJob t1 = new MyJob();
t1.start();
for (int i = 0; i < 100; i++) {
System.out.println("main:" + i);
}
}
}
class MyJob extends Thread{
@Override
public void run() {
for (int i = 0; i < 100; i++) {
System.out.println("MyJob:" + i);
}
}
}
2、实现Runnable接口 重写run方法
public class MiTest {
public static void main(String[] args) {
MyRunnable myRunnable = new MyRunnable();
Thread t1 = new Thread(myRunnable);
t1.start();
for (int i = 0; i < 1000; i++) {
System.out.println("main:" + i);
}
}
}
class MyRunnable implements Runnable{
@Override
public void run() {
for (int i = 0; i < 1000; i++) {
System.out.println("MyRunnable:" + i);
}
}
}
最常用的方式:
- 匿名内部类方式:
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 1000; i++) {
System.out.println("匿名内部类:" + i);
}
}
});
- lambda方式:
Thread t2 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
System.out.println("lambda:" + i);
}
});
3、实现Callable 重写call方法,配合FutureTask
Callable一般用于有返回结果的非阻塞的执行方法
同步非阻塞。
public class MiTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
//1. 创建MyCallable
MyCallable myCallable = new MyCallable();
//2. 创建FutureTask,传入Callable
FutureTask futureTask = new FutureTask(myCallable);
//3. 创建Thread线程
Thread t1 = new Thread(futureTask);
//4. 启动线程
t1.start();
//5. 做一些操作
//6. 要结果
Object count = futureTask.get();
System.out.println("总和为:" + count);
}
}
class MyCallable implements Callable{
@Override
public Object call() throws Exception {
int count = 0;
for (int i = 0; i < 100; i++) {
count += i;
}
return count;
}
}
如果在结果线程池使用:
public class MiTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
MyCallable myCallable = new MyCallable();
ExecutorService service = Executors.newCachedThreadPool();
Future<Integer> f = service.submit(myCallable);
Integer i = f.get();
}
}
class MyCallable implements Callable<Integer>{
@Override
public Integer call() throws Exception {
int count = 0;
for (int i = 0; i < 100; i++) {
count += i;
}
return count;
}
}
4、基于线程池构建线程
追其底层,其实只有一种,实现Runnble
三、线程状态
网上对线程状态的描述很多,有5种,6种,7种,都可以接受
5中状态一般是针对传统的线程状态来说(操作系统层面)
Java中给线程准备的6种状态
NEW:Thread对象被创建出来了,但是还没有执行start方法。
RUNNABLE:Thread对象调用了start方法,就为RUNNABLE状态(CPU调度/没有调度)
BLOCKED、WAITING、TIME_WAITING:都可以理解为是阻塞、等待状态,因为处在这三种状态下,CPU不会调度当前线程
BLOCKED:synchronized没有拿到同步锁,被阻塞的情况
WAITING:调用wait方法就会处于WAITING状态,需要被手动唤醒
TIME_WAITING:调用sleep方法或者join方法,会被自动唤醒,无需手动唤醒
TERMINATED:run方法执行完毕,线程生命周期到头了
在Java代码中验证一下效果
NEW:
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
});
System.out.println(t1.getState());
}
RUNNABLE:
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while(true){
}
});
t1.start();
Thread.sleep(500);
System.out.println(t1.getState());
}
BLOCKED:
public static void main(String[] args) throws InterruptedException {
Object obj = new Object();
Thread t1 = new Thread(() -> {
// t1线程拿不到锁资源,导致变为BLOCKED状态
synchronized (obj){
}
});
// main线程拿到obj的锁资源
synchronized (obj) {
t1.start();
Thread.sleep(500);
System.out.println(t1.getState());
}
}
WAITING:
public static void main(String[] args) throws InterruptedException {
Object obj = new Object();
Thread t1 = new Thread(() -> {
synchronized (obj){
try {
obj.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
t1.start();
Thread.sleep(500);
System.out.println(t1.getState());
}
TIMED_WAITING:
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
t1.start();
Thread.sleep(500);
System.out.println(t1.getState());
}
TERMINATED:
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
t1.start();
Thread.sleep(1000);
System.out.println(t1.getState());
}
四、线程的常用方法
1、获取当前线程
Thread的静态方法获取当前线程对象
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 获取当前线程的方法
Thread main = Thread.currentThread();
System.out.println(main);
// "Thread[" + getName() + "," + getPriority() + "," + group.getName() + "]";
// Thread[main,5,main]
}
2、线程的名字
在构建Thread对象完毕后,一定要设置一个有意义的名称,方面后期排查错误
public static void main(String[] args) throws ExecutionException, InterruptedException {
Thread t1 = new Thread(() -> {
System.out.println(Thread.currentThread().getName());
});
t1.setName("模块-功能-计数器");
t1.start();
}
3、线程的优先级
其实就是CPU调度线程的优先级、
java中给线程设置的优先级别有10个级别,从1~10任取一个整数。
如果超出这个范围,会排除参数异常的错误
public static void main(String[] args) throws ExecutionException, InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
System.out.println("t1:" + i);
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
System.out.println("t2:" + i);
}
});
t1.setPriority(1);
t2.setPriority(10);
t2.start();
t1.start();
}
4、线程的让步
可以通过Thread的静态方法yield,让当前线程从运行状态转变为就绪状态。
public static void main(String[] args) throws ExecutionException, InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
if(i == 50){
Thread.yield();
}
System.out.println("t1:" + i);
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
System.out.println("t2:" + i);
}
});
t2.start();
t1.start();
}
5、线程的休眠
Thread的静态方法,让线程从运行状态转变为等待状态
sleep有两个方法重载:
- 第一个就是native修饰的,让线程转为等待状态的效果
- 第二个是可以传入毫秒和一个纳秒的方法(如果纳秒值大于等于0.5毫秒,就给休眠的毫秒值+1。如果传入的毫秒值是0,纳秒值不为0,就休眠1毫秒)
sleep会抛出一个InterruptedException
public static void main(String[] args) throws InterruptedException {
System.out.println(System.currentTimeMillis());
Thread.sleep(1000);
System.out.println(System.currentTimeMillis());
}
6、线程的强占
Thread的非静态方法join方法
需要在某一个线程下去调用这个方法
如果在main线程中调用了t1.join(),那么main线程会进入到等待状态,需要等待t1线程全部执行完毕,在恢复到就绪状态等待CPU调度。
如果在main线程中调用了t1.join(2000),那么main线程会进入到等待状态,需要等待t1执行2s后,在恢复到就绪状态等待CPU调度。如果在等待期间,t1已经结束了,那么main线程自动变为就绪状态等待CPU调度。
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 10; i++) {
System.out.println("t1:" + i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
t1.start();
for (int i = 0; i < 10; i++) {
System.out.println("main:" + i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (i == 1){
try {
t1.join(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
7、守护线程
默认情况下,线程都是非守护线程
JVM会在程序中没有非守护线程时,结束掉当前JVM
主线程默认是非守护线程,如果主线程执行结束,需要查看当前JVM内是否还有非守护线程,如果没有JVM直接停止
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 10; i++) {
System.out.println("t1:" + i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
t1.setDaemon(true);
t1.start();
}
8、线程的等待和唤醒
(1)wait notify
可以让获取synchronized锁资源的线程通过wait方法进去到锁的等待池,并且会释放锁资源
可以让获取synchronized锁资源的线程,通过notify或者notifyAll方法,将等待池中的线程唤醒,添加到锁池中
notify随机的唤醒等待池中的一个线程到锁池
notifyAll将等待池中的全部线程都唤醒,并且添加到锁池
在调用wait方法和notify以及norifyAll方法时,必须在synchronized修饰的代码块或者方法内部才可以,因为要操作基于某个对象的锁的信息维护。
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
sync();
},"t1");
Thread t2 = new Thread(() -> {
sync();
},"t2");
t1.start();
t2.start();
Thread.sleep(12000);
synchronized (MiTest.class) {
MiTest.class.notifyAll();
}
}
public static synchronized void sync() {
try {
for (int i = 0; i < 10; i++) {
if(i == 5) {
MiTest.class.wait();
}
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName());
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
(2)LockSupport
在以前我们要阻塞和唤醒某一个具体的线程有很多限制比如:
1、因为wait()方法需要释放锁,所以必须在synchronized中使用,否则会抛出异常IllegalMonitorStateException
2、notify()方法也必须在synchronized中使用,并且应该指定对象
3、synchronized()、wait()、notify()对象必须一致,一个synchronized()代码块中只能有一个线程调用wait()或notify()
以上诸多限制,体现出了很多的不足,所以LockSupport的好处就体现出来了。
在JDK1.6中的java.util.concurrent的子包locks中引了LockSupport这个API,LockSupport是一个比较底层的工具类,用来创建锁和其他同步工具类的基本线程阻塞原语。java锁和同步器框架的核心 AQS: AbstractQueuedSynchronizer,就是通过调用 LockSupport .park()和 LockSupport .unpark()的方法,来实现线程的阻塞和唤醒的。我们先来看一个小程序:
public class T13_TestLockSupport {
public static void main(String[] args) {
//使用lombda表达式创建一个线程t
Thread t = new Thread(()->{
for (int i = 0; i < 10; i++) {
System.out.println(i);
if(i == 5) {
//使用LockSupport的park()方法阻塞当前线程t
LockSupport.park();
}
try {
//使当前线程t休眠1秒
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//启动当前线程t
t.start();
}
}
从以上的小程序中,我们不难看出LockSupport使用起来的是比较灵灵活的,没有了所谓的限制。我们来分析一下代码的执行过程,首先我们使用lombda表达式创建了线程对象 " t " ,然后通过 " t " 对象调用线程的启动方法start(),然后我们再看线程的内容,在for循环中,当 i 的值等于5的时候,我们调用了LockSupport的.park()方法使当前线程阻塞,注意看方法并没有加锁,就默认使当前线程阻塞了,由此可以看出LockSupprt.park()方法并没有加锁的限制。
我们再来看一个小程序:
public class T13_TestLockSupport {
public static void main(String[] args) {
//使用lombda表达式创建一个线程t
Thread t = new Thread(()->{
for (int i = 0; i < 10; i++) {
System.out.println(i);
if(i == 5) {
//使用LockSupport的park()方法阻塞当前线程t
LockSupport.park();
}
try {
//使当前线程t休眠1秒
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//启动当前线程t
t.start();
//唤醒线程t
LockSupport.unpark(t);
}
}
我们来分析一下以上小程序,我们只需要在第一个小程序的主线程中,调用LockSupport的unpark()方法,就可以唤醒某个具体的线程,这里我们指定了线程 " t " ,代码运行以后结果显而易见,线程并没有被阻塞,我们成功唤醒了线程 " t " ,在这里还有一点,需要我们来分析一下,在主线程中线程 " t " 调用了start()方法以后,因为紧接着执行了LockSupport的unpark()方法,所以也就是说,在线程 " t "还没有执行还没有被阻塞的时候,已经调用了LockSupport的unpark()方法来唤醒线程 " t " ,之后线程 " t "才启动调用了LockSupport的park()来使线程 " t " 阻塞,但是线程 " t " 并没有被阻塞,由此可以看出,LockSupport的unpark()方法可以先于LockSupport的park()方法执行。
我们再来看最后一个小程序:
public class T13_TestLockSupport {
public static void main(String[] args) {
//使用lombda表达式创建一个线程t
Thread t = new Thread(()->{
for (int i = 0; i < 10; i++) {
System.out.println(i);
if(i == 5) {
//调用LockSupport的park()方法阻塞当前线程t
LockSupport.park();
}
if(i == 8){
//调用LockSupport的park()方法阻塞当前线程t
LockSupport.park();
}
try {
//使当前线程t休眠1秒
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
//启动当前线程t
t.start();
//唤醒线程t
LockSupport.unpark(t);
}
}
我们来分析一下以上小程序,在第二个小程序的基础上又添加了一个if判断,在i等于8的时候再次调用LockSupport的park()方法来使线程 " t " 阻塞, 我们可以看到线程被阻塞了,原因是LockSupport的unpark()方法就像是获得了一个“令牌”,而LockSupport的park()方法就像是在识别“令牌”,当主线程调用了LockSupport.unpark(t)方法也就说明线程 " t " 已经获得了”令牌”,当线程 " t " 再调用LockSupport的park()方法时,线程 " t " 已经有令牌了,这样他就会马上再继续运行,也就不会被阻塞了,但是当i等于8的时候线程 " t " 再次调用了LockSupport的park()方法使线程再次进入阻塞状态,这个时候“令牌”已经被使用作废掉了,也就无法阻塞线程 " t " 了,而且如果主线程处于等待“令牌”状态时,线程 " t " 再次调用了LockSupport的park()方法,那么线程 " t "就会永远阻塞下去,即使调用unpark()方法也无法唤醒了。
由以上三个小程序我们可以总结得出以下几点:
1、LockSupport不需要synchornized加锁就可以实现线程的阻塞和唤醒
2、LockSupport.unpartk()可以先于LockSupport.park()执行,并且线程不会阻塞
3、如果一个线程处于等待状态,连续调用了两次park()方法,就会使该线程永远无法被唤醒
LockSupport中park()和unpark()方法的实现原理
park()和unpark()方法的实现是由Unsefa类提供的,而Unsefa类是由C和C++语言完成的,其实原理也是比较好理解的,它主要通过一个变量作为一个标识,变量值在0,1之间来回切换,当这个变量大于0的时候线程就获得了“令牌”,从这一点我们不难知道,其实park()和unpark()方法就是在改变这个变量的值,来达到线程的阻塞和唤醒的,具体实现不做赘述了。
五、线程打断interrupt
- t.interrupt() 打断某个线程(设置标志位)
- t.isInterrupted() 查询某线程是否被打断过(查询标志位)
- Thread.interrupted() 查询当前线程是否被打断过,并重置打断标志。
sleep()、wait()、join() 时被打断,会抛出InterruptedException 异常,并且catch 异常后自动重置打断标志。
interrupt 是没法打断线程抢synchronized 和 lock 锁的。想要打断抢锁的线程,则用lock.lockInterruptibly() (可以被打断)上锁,然后对该线程interrupt 设标志位,则会抛InterruptedException 异常。
可以利用interrupt 优雅的停止线程。
六、线程结束
如何优雅的结束一个线程?
e.g.上传一个大文件,正在处理费时的计算,如何优雅的结束这个线程?
stop方法(不用)
强制让线程结束,无论你在干嘛,不推荐使用当然当然方式,但是,他确实可以把线程干掉
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
t1.start();
Thread.sleep(500);
t1.stop();
System.out.println(t1.getState());
}
为什么不建议使用stop 方法?
太粗暴了,stop 方法不管线程处于什么状态,它会直接停止线程,并释放所有锁,但是不会做善后的工作,很容易产生数据不一致问题。
suspend 暂停线程和 resume 恢复线程(不用)
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
t1.start();
Thread.sleep(500);
t1.suspend();
Thread.sleep(500);
t1.resume();
}
这两个方法是配套使用,也被废止了,原因和stop 类似,也容易产生数据不一致问题。还有如调用了suspend 暂停线程后,并不会释放锁,然后忘记调resume 方法,这把锁永远不会被释放,容易产生死锁。
使用共享变量(很少会用)
这种方式用的也不多,有的线程可能会通过死循环来保证一直运行。
咱们可以通过修改共享变量在破坏死循环,让线程退出循环,结束run方法
static volatile boolean flag = true;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while(flag){
// 处理任务
}
System.out.println("任务结束");
});
t1.start();
Thread.sleep(500);
flag = false;
}
interrupt方式
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while(!Thread.currentThread().isInterrupted()){
// 处理业务
}
System.out.println("t1结束");
});
t1.start();
Thread.sleep(500);
t1.interrupt();
}
通过打断WAITING或者TIMED_WAITING状态的线程,从而抛出异常自行处理
这种停止线程方式是最常用的一种,在框架和JUC中也是最常见的
第二章、并发编程的三大特性
一、可见性(visibility)
1、什么是可见性
可见性问题是基于CPU位置出现的,CPU处理速度非常快,相对CPU来说,去主内存获取数据这个事情太慢了,CPU就提供了L1,L2,L3的三级缓存,每次去主内存拿完数据后,就会存储到CPU的三级缓存,每次去三级缓存拿数据,效率肯定会提升。
这就带来了问题,现在CPU都是多核,每个线程的工作内存(CPU三级缓存)都是独立的,会告知每个线程中做修改时,只改自己的工作内存,没有及时的同步到主内存,导致数据不一致问题。
可见性问题的代码逻辑
private static boolean flag = true;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while (flag) {
// ....
}
System.out.println("t1线程结束");
});
t1.start();
Thread.sleep(10);
flag = false;
System.out.println("主线程将flag改为false");
}
2、解决可见性的方式
(1)volatile
volatile是一个关键字,用来修饰成员变量。
如果属性被volatile修饰,相当于会告诉CPU,对当前属性的操作,不允许使用CPU的缓存,必须去和主内存操作
volatile的内存语义:
- volatile属性被写:当写一个volatile变量,JMM会将当前线程对应的CPU缓存及时的刷新到主内存中
- volatile属性被读:当读一个volatile变量,JMM会将对应的CPU缓存中的内存设置为无效,必须去主内存中重新读取共享变量
其实加了volatile就是告知CPU,对当前属性的读写操作,不允许使用CPU缓存,加了volatile修饰的属性,会在转为汇编之后后,追加一个lock的前缀,CPU执行这个指令时,如果带有lock前缀会做两个事情:
- 将当前处理器缓存行的数据写回到主内存
- 这个写回的数据,在其他的CPU内核的缓存中,直接无效。
总结:volatile就是让CPU每次操作这个数据时,必须立即同步到主内存,以及从主内存读取数据。
private volatile static boolean flag = true;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while (flag) {
// ....
}
System.out.println("t1线程结束");
});
t1.start();
Thread.sleep(10);
flag = false;
System.out.println("主线程将flag改为false");
}
(2)synchronized
synchronized也是可以解决可见性问题的,synchronized的内存语义。
如果涉及到了synchronized的同步代码块或者是同步方法,获取锁资源之后,将内部涉及到的变量从CPU缓存中移除,必须去主内存中重新拿数据,而且在释放锁之后,会立即将CPU缓存中的数据同步到主内存。
private static boolean flag = true;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while (flag) {
synchronized (MiTest.class){
//...
}
System.out.println(111);
}
System.out.println("t1线程结束");
});
t1.start();
Thread.sleep(10);
flag = false;
System.out.println("主线程将flag改为false");
}
(3)Lock
Lock锁保证可见性的方式和synchronized完全不同,synchronized基于他的内存语义,在获取锁和释放锁时,对CPU缓存做一个同步到主内存的操作。
Lock锁是基于volatile实现的。Lock锁内部再进行加锁和释放锁时,会对一个由volatile修饰的state属性进行加减操作。
如果对volatile修饰的属性进行写操作,CPU会执行带有lock前缀的指令,CPU会将修改的数据,从CPU缓存立即同步到主内存,同时也会将其他的属性也立即同步到主内存中。还会将其他CPU缓存行中的这个数据设置为无效,必须重新从主内存中拉取。
private static boolean flag = true;
private static Lock lock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
while (flag) {
lock.lock();
try{
//...
}finally {
lock.unlock();
}
}
System.out.println("t1线程结束");
});
t1.start();
Thread.sleep(10);
flag = false;
System.out.println("主线程将flag改为false");
}
(4)final
final修饰的属性,在运行期间是不允许修改的,这样一来,就间接的保证了可见性,所有多线程读取final属性,值肯定是一样。
final并不是说每次取数据从主内存读取,他没有这个必要,而且final和volatile是不允许同时修饰一个属性的
final修饰的内容已经不允许再次被写了,而volatile是保证每次读写数据去主内存读取,并且volatile会影响一定的性能,就不需要同时修饰。
二、有序性(ordering)
1、什么是有序性
在Java中,.java文件中的内容会被编译,在执行前需要再次转为CPU可以识别的指令,CPU在执行这些指令时,为了提升执行效率,在不影响最终结果的前提下(满足一些要求),会对指令进行重排。
指令乱序执行的原因,是为了尽可能的发挥CPU的性能。
Java中的程序是乱序执行的。
Java程序验证乱序执行效果:
static int a,b,x,y;
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < Integer.MAX_VALUE; i++) {
a = 0;
b = 0;
x = 0;
y = 0;
Thread t1 = new Thread(() -> {
a = 1;
x = b;
});
Thread t2 = new Thread(() -> {
b = 1;
y = a;
});
t1.start();
t2.start();
t1.join();
t2.join();
if(x == 0 && y == 0){
System.out.println("第" + i + "次,x = "+ x + ",y = " + y);
}
}
}
单例模式由于指令重排序可能会出现问题:
线程可能会拿到没有初始化的对象,导致在使用时,可能由于内部属性为默认值,导致出现一些不必要的问题
private static volatile MiTest test;
private MiTest(){}
public static MiTest getInstance(){
// B
if(test == null){
synchronized (MiTest.class){
if(test == null){
// A , 开辟空间,test指向地址,初始化
test = new MiTest();
}
}
}
return test;
}
2、as-if-serial
as-if-serial语义:
不论指定如何重排序,需要保证单线程的程序执行结果是不变的。
而且如果存在依赖的关系,那么也不可以做指令重排。
// 这种情况肯定不能做指令重排序
int i = 0;
i++;
// 这种情况肯定不能做指令重排序
int j = 200;
j * 100;
j + 100;
// 这里即便出现了指令重排,也不可以影响最终的结果,20100
3、happens-before
具体规则:
- 单线程happen-before原则:在同一个线程中,书写在前面的操作happen-before后面的操作。
- 锁的happen-before原则:同一个锁的unlock操作happen-before此锁的lock操作。
- volatile的happen-before原则: 对一个volatile变量的写操作happen-before对此变量的任意操作。
- happen-before的传递性原则: 如果A操作 happen-before B操作,B操作happen-before C操作,那么A操作happen-before C操作。
- 线程启动的happen-before原则:同一个线程的start方法happen-before此线程的其它方法。
- 线程中断的happen-before原则:对线程interrupt方法的调用happen-before被中断线程的检测到中断发送的代码。
- 线程终结的happen-before原则:线程中的所有操作都happen-before线程的终止检测。
- 对象创建的happen-before原则:一个对象的初始化完成先于他的finalize方法调用。
JMM只有在不出现上述8中情况时,才不会触发指令重排效果。
不需要过分的关注happens-before原则,只需要可以写出线程安全的代码就可以了。
4、volatile
如果需要让程序对某一个属性的操作不出现指令重排,除了满足happens-before原则之外,还可以基于volatile修饰属性,从而对这个属性的操作,就不会出现指令重排的问题了。
volatile如何实现的禁止指令重排?
内存屏障概念。将内存屏障看成一条指令。
会在两个操作之间,添加上一道指令,这个指令就可以避免上下执行的其他指令进行重排序。
三、原子性(atomicity)
1、什么是并发编程的原子性
JMM(Java Memory Model)。不同的硬件和不同的操作系统在内存上的操作有一定差异的。Java为了解决相同代码在不同操作系统上出现的各种问题,用JMM屏蔽掉各种硬件和操作系统带来的差异。
让Java的并发编程可以做到跨平台。
JMM规定所有变量都会存储在主内存中,在操作的时候,需要从主内存中复制一份到线程内存(CPU内存),在线程内部做计算。然后再写回主内存中(不一定!)。
原子性的定义:原子性指一个操作是不可分割的,不可中断的,一个线程在执行时,另一个线程不会影响到他。
并发编程的原子性用代码阐述:
private static int count;
public static void increment(){
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
count++;
}
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
increment();
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
increment();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(count);
}
当前程序:多线程操作共享数据时,预期的结果,与最终的结果不符。
原子性:多线程操作临界资源,预期的结果与最终结果一致。
通过对这个程序的分析,可以查看出,++的操作,一共分为了三部,首先是线程从主内存拿到数据保存到CPU的寄存器中,然后在寄存器中进行+1操作,最终将结果写回到主内存当中。
2、保证并发编程的原子性
(1)synchronized
因为++操作可以从指令中查看到
可以在方法上追加synchronized关键字或者采用同步代码块的形式来保证原子性
synchronized可以让避免多线程同时操作临街资源,同一时间点,只会有一个线程正在操作临界资源
(2)CAS
到底什么是CAS
compare and swap也就是比较和交换,他是一条CPU的并发原语。
他在替换内存的某个位置的值时,首先查看内存中的值与预期值是否一致,如果一致,执行替换操作。这个操作是一个原子性操作。
Java中基于Unsafe的类提供了对CAS的操作的方法,JVM会帮助我们将方法实现CAS汇编指令。
但是要清楚CAS只是比较和交换,在获取原值的这个操作上,需要你自己实现。
private static AtomicInteger count = new AtomicInteger(0);
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
count.incrementAndGet();
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
count.incrementAndGet();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(count);
}
Doug Lea在CAS的基础上帮助我们实现了一些原子类,其中就包括现在看到的AtomicInteger,还有其他很多原子类……
CAS的缺点:CAS只能保证对一个变量的操作是原子性的,无法实现对多行代码实现原子性。
CAS的问题:
- ABA问题:问题如下,可以引入版本号的方式,来解决ABA的问题。Java中提供了一个类在CAS时,针对各个版本追加版本号的操作。 AtomicStampeReference
- AtomicStampedReference在CAS时,不但会判断原值,还会比较版本信息。
public static void main(String[] args) {
AtomicStampedReference<String> reference = new AtomicStampedReference<>("AAA",1);
String oldValue = reference.getReference();
int oldVersion = reference.getStamp();
boolean b = reference.compareAndSet(oldValue, "B", oldVersion, oldVersion + 1);
System.out.println("修改1版本的:" + b);
boolean c = reference.compareAndSet("B", "C", 1, 1 + 1);
System.out.println("修改2版本的:" + c);
}
-
自旋时间过长问题:
- 可以指定CAS一共循环多少次,如果超过这个次数,直接失败/或者挂起线程。(自旋锁、自适应自旋锁)
- 可以在CAS一次失败后,将这个操作暂存起来,后面需要获取结果时,将暂存的操作全部执行,再返回最后的结果。
(3)Lock锁
Lock锁是在JDK1.5由Doug Lea研发的,他的性能相比synchronized在JDK1.5的时期,性能好了很多多,但是在JDK1.6对synchronized优化之后,性能相差不大,但是如果涉及并发比较多时,推荐ReentrantLock锁,性能会更好。
实现方式:
private static int count;
private static ReentrantLock lock = new ReentrantLock();
public static void increment() {
lock.lock();
try {
count++;
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
increment();
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 100; i++) {
increment();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(count);
}
ReentrantLock可以直接对比synchronized,在功能上来说,都是锁。
但是ReentrantLock的功能性相比synchronized更丰富。
ReentrantLock底层是基于AQS实现的,有一个基于CAS维护的state变量来实现锁的操作。
(4)ThreadLocal
Java中的四种引用类型
Java中的使用引用类型分别是强,软,弱,虚。
User user = new User();
在 Java 中最常见的就是强引用,把一个对象赋给一个引用变量,这个引用变量就是一个强引用。当一个对象被强引用变量引用时,它始终处于可达状态,它是不可能被垃圾回收机制回收的,即使该对象以后永远都不会被用到 JVM 也不会回收。因此强引用是造成 Java 内存泄漏的主要原因之一。
SoftReference
其次是软引用,对于只有软引用的对象来说,当系统内存足够时它不会被回收,当系统内存空间不足时它会被回收。软引用通常用在对内存敏感的程序中,作为缓存使用。
然后是弱引用,它比软引用的生存期更短,对于只有弱引用的对象来说,只要垃圾回收机制一运行,不管 JVM 的内存空间是否足够,总会回收该对象占用的内存。可以解决内存泄漏问题,ThreadLocal就是基于弱引用解决内存泄漏的问题。
最后是虚引用,它不能单独使用,必须和引用队列联合使用。虚引用的主要作用是跟踪对象被垃圾回收的状态。不过在开发中,我们用的更多的还是强引用。
ThreadLocal保证原子性的方式,是不让多线程去操作临界资源,让每个线程去操作属于自己的数据
代码实现
static ThreadLocal tl1 = new ThreadLocal();
static ThreadLocal tl2 = new ThreadLocal();
public static void main(String[] args) {
tl1.set("123");
tl2.set("456");
Thread t1 = new Thread(() -> {
System.out.println("t1:" + tl1.get());
System.out.println("t1:" + tl2.get());
});
t1.start();
System.out.println("main:" + tl1.get());
System.out.println("main:" + tl2.get());
}
ThreadLocal实现原理:
- 每个Thread中都存储着一个成员变量,ThreadLocalMap
- ThreadLocal本身不存储数据,像是一个工具类,基于ThreadLocal去操作ThreadLocalMap
- ThreadLocalMap本身就是基于Entry[]实现的,因为一个线程可以绑定多个ThreadLocal,这样一来,可能需要存储多个数据,所以采用Entry[]的形式实现。
- 每一个现有都自己独立的ThreadLocalMap,再基于ThreadLocal对象本身作为key,对value进行存取
- ThreadLocalMap的key是一个弱引用,弱引用的特点是,即便有弱引用,在GC时,也必须被回收。这里是为了在ThreadLocal对象失去引用后,如果key的引用是强引用,会导致ThreadLocal对象无法被回收
ThreadLocal内存泄漏问题:
- 如果ThreadLocal引用丢失,key因为弱引用会被GC回收掉,如果同时线程还没有被回收,就会导致内存泄漏,内存中的value无法被回收,同时也无法被获取到。
- 只需要在使用完毕ThreadLocal对象之后,及时的调用remove方法,移除Entry即可
第三章、锁
一、锁的分类
1、可重入锁、不可重入锁
Java中提供的synchronized,ReentrantLock,ReentrantReadWriteLock都是可重入锁。
重入:当前线程获取到A锁,在获取之后尝试再次获取A锁是可以直接拿到的。
不可重入:当前线程获取到A锁,在获取之后尝试再次获取A锁,无法获取到的,因为A锁被当前线程占用着,需要等待自己释放锁再获取锁。
2、乐观锁、悲观锁
Java中提供的synchronized,ReentrantLock,ReentrantReadWriteLock都是悲观锁。
Java中提供的CAS操作,就是乐观锁的一种实现。
悲观锁:获取不到锁资源时,会将当前线程挂起(进入BLOCKED、WAITING),线程挂起会涉及到用户态和内核的太的切换,而这种切换是比较消耗资源的。
- 用户态:JVM可以自行执行的指令,不需要借助操作系统执行。
- 内核态:JVM不可以自行执行,需要操作系统才可以执行。
乐观锁:获取不到锁资源,可以再次让CPU调度,重新尝试获取锁资源。
Atomic原子性类中,就是基于CAS乐观锁实现的。
3、公平锁、非公平锁
Java中提供的synchronized只能是非公平锁。
Java中提供的ReentrantLock,ReentrantReadWriteLock可以实现公平锁和非公平锁
公平锁:线程A获取到了锁资源,线程B没有拿到,线程B去排队,线程C来了,锁被A持有,同时线程B在排队。直接排到B的后面,等待B拿到锁资源或者是B取消后,才可以尝试去竞争锁资源。
非公平锁:线程A获取到了锁资源,线程B没有拿到,线程B去排队,线程C来了,先尝试竞争一波
- 拿到锁资源:开心,插队成功。
- 没有拿到锁资源:依然要排到B的后面,等待B拿到锁资源或者是B取消后,才可以尝试去竞争锁资源。
4、互斥锁、共享锁
Java中提供的synchronized、ReentrantLock是互斥锁。
Java中提供的ReentrantReadWriteLock,有互斥锁也有共享锁。
互斥锁:同一时间点,只会有一个线程持有者当前互斥锁。
共享锁:同一时间点,当前共享锁可以被多个线程同时持有。
二、深入synchronized
1、类锁、对象锁
synchronized的使用一般就是同步方法和同步代码块。
synchronized的锁是基于对象实现的。
如果使用同步方法
- static:此时使用的是当前类.class作为锁(类锁)
- 非static:此时使用的是当前对象做为锁(对象锁)
public class MiTest {
public static void main(String[] args) {
// 锁的是,当前Test.class
Test.a();
Test test = new Test();
// 锁的是new出来的test对象
test.b();
}
}
class Test{
public static synchronized void a(){
System.out.println("1111");
}
public synchronized void b(){
System.out.println("2222");
}
}
2、synchronized 的优化
在JDK1.5的时候,Doug Lee推出了ReentrantLock,lock的性能远高于synchronized,所以JDK团队就在JDK1.6中,对synchronized做了大量的优化。
锁消除:在synchronized修饰的代码中,如果不存在操作临界资源的情况,会触发锁消除,你即便写了synchronized,他也不会触发。
public synchronized void method(){
// 没有操作临界资源
// 此时这个方法的synchronized你可以认为木有~~
}
锁膨胀:如果在一个循环中,频繁的获取和释放做资源,这样带来的消耗很大,锁膨胀就是将锁的范围扩大,避免频繁的竞争和获取锁资源带来不必要的消耗。
public void method(){
for(int i = 0;i < 999999;i++){
synchronized(对象){
}
}
// 这是上面的代码会触发锁膨胀
synchronized(对象){
for(int i = 0;i < 999999;i++){
}
}
}
锁升级:ReentrantLock的实现,是先基于乐观锁的CAS尝试获取锁资源,如果拿不到锁资源,才会挂起线程。synchronized在JDK1.6之前,完全就是获取不到锁,立即挂起当前线程,所以synchronized性能比较差。
synchronized就在JDK1.6做了锁升级的优化
- 无锁、匿名偏向:当前对象没有作为锁存在。
- 偏向锁:如果当前锁资源,只有一个线程在频繁的获取和释放,那么这个线程过来,只需要判断,当前指向的线程是否是当前线程 。
- 如果是,直接拿着锁资源走。
- 如果当前线程不是我,基于CAS的方式,尝试将偏向锁指向当前线程。如果获取不到,触发锁升级,升级为轻量级锁。(偏向锁状态出现了锁竞争的情况)
- 轻量级锁:会采用自旋锁的方式去频繁的以CAS的形式获取锁资源(采用的是自适应自旋锁)
- 如果成功获取到,拿着锁资源走
- 如果自旋了一定次数,没拿到锁资源,锁升级。
- 重量级锁:就是最传统的synchronized方式,拿不到锁资源,就挂起当前线程。(用户态&内核态)
3、synchronized 实现原理
synchronized是基于对象实现的。
先要对Java中对象在堆内存的存储有一个了解。
展开MarkWord
MarkWord中标记着四种锁的信息:无锁、偏向锁、轻量级锁、重量级锁。
4、synchronized 的锁升级
为了可以在Java中看到对象头的MarkWord信息,需要导入依赖
<dependency>
<groupId>org.openjdk.jol</groupId>
<artifactId>jol-core</artifactId>
<version>0.9</version>
</dependency>
锁默认情况下,开启了偏向锁延迟。
偏向锁在升级为轻量级锁时,会涉及到偏向锁撤销,需要等到一个安全点(STW),才可以做偏向锁撤销,在明知道有并发情况,就可以选择不开启偏向锁,或者是设置偏向锁延迟开启
因为JVM在启动时,需要加载大量的.class文件到内存中,这个操作会涉及到synchronized的使用,为了避免出现偏向锁撤销操作,JVM启动初期,有一个延迟4s开启偏向锁的操作
如果正常开启偏向锁了,那么不会出现无锁状态,对象会直接变为匿名偏向
public static void main(String[] args) throws InterruptedException {
Thread.sleep(5000);
Object o = new Object();
System.out.println(ClassLayout.parseInstance(o).toPrintable());
new Thread(() -> {
synchronized (o){
//t1 - 偏向锁
System.out.println("t1:" + ClassLayout.parseInstance(o).toPrintable());
}
}).start();
//main - 偏向锁 - 轻量级锁CAS - 重量级锁
synchronized (o){
System.out.println("main:" + ClassLayout.parseInstance(o).toPrintable());
}
}
整个锁升级状态的转变:
Lock Record以及ObjectMonitor存储的内容
5、重量锁底层ObjectMonitor
需要去找到openjdk,在百度中直接搜索openjdk,第一个链接就是
找到ObjectMonitor的两个文件,hpp,cpp
先查看核心属性:http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/69087d08d473/src/share/vm/runtime/objectMonitor.hpp
ObjectMonitor() {
_header = NULL; // header存储着MarkWord
_count = 0; // 竞争锁的线程个数
_waiters = 0, // wait的线程个数
_recursions = 0; // 标识当前synchronized锁重入的次数
_object = NULL;
_owner = NULL; // 持有锁的线程
_WaitSet = NULL; // 保存wait的线程信息,双向链表
_WaitSetLock = 0 ;
_Responsible = NULL ;
_succ = NULL ;
_cxq = NULL ; // 获取锁资源失败后,线程要放到当前的单向链表中
FreeNext = NULL ;
_EntryList = NULL ; // _cxq以及被唤醒的WaitSet中的线程,在一定机制下,会放到EntryList中
_SpinFreq = 0 ;
_SpinClock = 0 ;
OwnerIsThread = 0 ;
_previous_owner_tid = 0;
}
适当的查看几个C++中实现的加锁流程
http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/69087d08d473/src/share/vm/runtime/objectMonitor.cpp
TryLock
int ObjectMonitor::TryLock (Thread * Self) {
for (;;) {
// 拿到持有锁的线程
void * own = _owner ;
// 如果有线程持有锁,告辞
if (own != NULL) return 0 ;
// 说明没有线程持有锁,own是null,cmpxchg指令就是底层的CAS实现。
if (Atomic::cmpxchg_ptr (Self, &_owner, NULL) == NULL) {
// 成功获取锁资源
return 1 ;
}
// 这里其实重试操作没什么意义,直接返回-1
if (true) return -1 ;
}
}
try_entry
bool ObjectMonitor::try_enter(Thread* THREAD) {
// 在判断_owner是不是当前线程
if (THREAD != _owner) {
// 判断当前持有锁的线程是否是当前线程,说明轻量级锁刚刚升级过来的情况
if (THREAD->is_lock_owned ((address)_owner)) {
_owner = THREAD ;
_recursions = 1 ;
OwnerIsThread = 1 ;
return true;
}
// CAS操作,尝试获取锁资源
if (Atomic::cmpxchg_ptr (THREAD, &_owner, NULL) != NULL) {
// 没拿到锁资源,告辞
return false;
}
// 拿到锁资源
return true;
} else {
// 将_recursions + 1,代表锁重入操作。
_recursions++;
return true;
}
}
enter(想方设法拿到锁资源,如果没拿到,挂起扔到_cxq单向链表中)
void ATTR ObjectMonitor::enter(TRAPS) {
// 拿到当前线程
Thread * const Self = THREAD ;
void * cur ;
// CAS走你,
cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
if (cur == NULL) {
// 拿锁成功
return ;
}
// 锁重入操作
if (cur == Self) {
// TODO-FIXME: check for integer overflow! BUGID 6557169.
_recursions ++ ;
return ;
}
//轻量级锁过来的。
if (Self->is_lock_owned ((address)cur)) {
_recursions = 1 ;
_owner = Self ;
OwnerIsThread = 1 ;
return ;
}
// 走到这了,没拿到锁资源,count++
Atomic::inc_ptr(&_count);
for (;;) {
jt->set_suspend_equivalent();
// 入队操作,进到cxq中
EnterI (THREAD) ;
if (!ExitSuspendEquivalent(jt)) break ;
_recursions = 0 ;
_succ = NULL ;
exit (false, Self) ;
jt->java_suspend_self();
}
}
// count--
Atomic::dec_ptr(&_count);
}
EnterI
for (;;) {
// 入队
node._next = nxt = _cxq ;
// CAS的方式入队。
if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;
// 重新尝试获取锁资源
if (TryLock (Self) > 0) {
assert (_succ != Self , "invariant") ;
assert (_owner == Self , "invariant") ;
assert (_Responsible != Self , "invariant") ;
return ;
}
}
三、深入ReentrantLock
1、ReentrantLock和synchronized的区别
核心区别:
- ReentrantLock 是个类,synchronized 是关键字,当然都是在JVM 层面实现互斥锁的方式
效率区别:
- 如果竞争比较激烈,推荐ReentrantLock去实现,不存在锁升级概念。而synchronized是存在锁升级概念的,如果升级到重量级锁,是不存在锁降级的。
底层实现区别:
- 实现原理是不一样,ReentrantLock基于AQS实现的,synchronized是基于ObjectMonitor
功能向的区别:
- ReentrantLock的功能比synchronized更全面。
- ReentrantLock支持公平锁和非公平锁
- ReentrantLock可以指定等待锁资源的时间。
选择哪个:如果你对并发编程特别熟练,推荐使用ReentrantLock,功能更丰富。如果掌握的一般般,使用synchronized会更好
2、AQS概述
AQS就是AbstractQueuedSynchronizer抽象类,AQS其实就是JUC包下的一个基类,JUC下的很多内容都是基于AQS实现了部分功能,比如ReentrantLock,ThreadPoolExecutor,阻塞队列,CountDownLatch,Semaphore,CyclicBarrier等等都是基于AQS实现。
首先AQS中提供了一个由volatile修饰,并且采用CAS方式修改的int类型的state变量。
其次AQS中维护了一个双向链表,有head,有tail,并且每个节点都是Node对象
static final class Node {
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
}
AQS内部结构和属性
3、加锁流程源码剖析
(1)加锁流程概述
这个是非公平锁的流程
(2)三种加锁源码分析
lock方法
-
执行lock方法后,公平锁和非公平锁的执行套路不一样
// 非公平锁 final void lock() { // 上来就先基于CAS的方式,尝试将state从0改为1 if (compareAndSetState(0, 1)) // 获取锁资源成功,会将当前线程设置到exclusiveOwnerThread属性,代表是当前线程持有着锁资源 setExclusiveOwnerThread(Thread.currentThread()); else // 执行acquire,尝试获取锁资源 acquire(1); } // 公平锁 final void lock() { // 执行acquire,尝试获取锁资源 acquire(1); }
-
acquire方法,是公平锁和非公平锁的逻辑一样
public final void acquire(int arg) { // tryAcquire:再次查看,当前线程是否可以尝试获取锁资源 if (!tryAcquire(arg) && // 没有拿到锁资源 // addWaiter(Node.EXCLUSIVE):将当前线程封装为Node节点,插入到AQS的双向链表的结尾 // acquireQueued:查看我是否是第一个排队的节点,如果是可以再次尝试获取锁资源,如果长时间拿不到,挂起线程 // 如果不是第一个排队的额节点,就尝试挂起线程即可 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 中断线程的操作 selfInterrupt(); }
-
tryAcquire方法竞争锁最资源的逻辑,分为公平锁和非公平锁
// 非公平锁实现 final boolean nonfairTryAcquire(int acquires) { // 获取当前线程 final Thread current = Thread.currentThread(); // 获取了state熟属性 int c = getState(); // 判断state当前是否为0,之前持有锁的线程释放了锁资源 if (c == 0) { // 再次抢一波锁资源 if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); // 拿锁成功返回true return true; } } // 不是0,有线程持有着锁资源,如果是,证明是锁重入操作 else if (current == getExclusiveOwnerThread()) { // 将state + 1 int nextc = c + acquires; if (nextc < 0) // 说明对重入次数+1后,超过了int正数的取值范围 // 01111111 11111111 11111111 11111111 // 10000000 00000000 00000000 00000000 // 说明重入的次数超过界限了。 throw new Error("Maximum lock count exceeded"); // 正常的将计算结果,复制给state setState(nextc); // 锁重入成功 return true; } // 返回false return false; } // 公平锁实现 protected final boolean tryAcquire(int acquires) { // 获取当前线程 final Thread current = Thread.currentThread(); // .... int c = getState(); if (c == 0) { // 查看AQS中是否有排队的Node // 没人排队抢一手 。有人排队,如果我是第一个,也抢一手 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的双向队列中排队 // 返回false,代表没人排队 public final boolean hasQueuedPredecessors() { // 头尾节点 Node t = tail; Node h = head; // s为头结点的next节点 Node s; // 如果头尾节点相等,证明没有线程排队,直接去抢占锁资源 return h != t && // s节点不为null,并且s节点的线程为当前线程(排在第一名的是不是我) (s == null || s.thread != Thread.currentThread()); }
-
addWaite方法,将没有拿到锁资源的线程扔到AQS队列中去排队
// 没有拿到锁资源,过来排队, mode:代表互斥锁 private Node addWaiter(Node mode) { // 将当前线程封装为Node, Node node = new Node(Thread.currentThread(), mode); // 拿到尾结点 Node pred = tail; // 如果尾结点不为null if (pred != null) { // 当前节点的prev指向尾结点 node.prev = pred; // 以CAS的方式,将当前线程设置为tail节点 if (compareAndSetTail(pred, node)) { // 将之前的尾结点的next指向当前节点 pred.next = node; return node; } } // 如果CAS失败,以死循环的方式,保证当前线程的Node一定可以放到AQS队列的末尾 enq(node); return node; } private Node enq(final Node node) { for (;;) { // 拿到尾结点 Node t = tail; // 如果尾结点为空,AQS中一个节点都没有,构建一个伪节点,作为head和tail if (t == null) { if (compareAndSetHead(new Node())) tail = head; } else { // 比较熟悉了,以CAS的方式,在AQS中有节点后,插入到AQS队列的末尾 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
-
acquireQueued方法,判断当前线程是否还能再次尝试获取锁资源,如果不能再次获取锁资源,或者又没获取到,尝试将当前线程挂起
// 当前没有拿到锁资源后,并且到AQS排队了之后触发的方法。 中断操作这里不用考虑 final boolean acquireQueued(final Node node, int arg) { // 不考虑中断 // failed:获取锁资源是否失败(这里简单掌握落地,真正触发的,还是tryLock和lockInterruptibly) boolean failed = true; try { boolean interrupted = false; // 死循环………… for (;;) { // 拿到当前节点的前继节点 final Node p = node.predecessor(); // 前继节点是否是head,如果是head,再次执行tryAcquire尝试获取锁资源。 if (p == head && tryAcquire(arg)) { // 获取锁资源成功 // 设置头结点为当前获取锁资源成功Node,并且取消thread信息 setHead(node); // help GC p.next = null; // 获取锁失败标识为false failed = false; return interrupted; } // 没拿到锁资源…… // shouldParkAfterFailedAcquire:基于上一个节点转改来判断当前节点是否能够挂起线程,如果可以返回true, // 如果不能,就返回false,继续下次循环 if (shouldParkAfterFailedAcquire(p, node) && // 这里基于Unsafe类的park方法,将当前线程挂起 parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) // 在lock方法中,基本不会执行。 cancelAcquire(node); } } // 获取锁资源成功后,先执行setHead private void setHead(Node node) { // 当前节点作为头结点 伪 head = node; // 头结点不需要线程信息 node.thread = null; node.prev = null; } // 当前Node没有拿到锁资源,或者没有资格竞争锁资源,看一下能否挂起当前线程 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { // -1,SIGNAL状态:代表当前节点的后继节点,可以挂起线程,后续我会唤醒我的后继节点 // 1,CANCELLED状态:代表当前节点以及取消了 int ws = pred.waitStatus; if (ws == Node.SIGNAL) // 上一个节点为-1之后,当前节点才可以安心的挂起线程 return true; if (ws > 0) { // 如果当前节点的上一个节点是取消状态,我需要往前找到一个状态不为1的Node,作为他的next节点 // 找到状态不为1的节点后,设置一下next和prev do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { // 上一个节点的状态不是1或者-1,那就代表节点状态正常,将上一个节点的状态改为-1 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
tryLock方法
-
tryLock();
// tryLock方法,无论公平锁还有非公平锁。都会走非公平锁抢占锁资源的操作 // 就是拿到state的值, 如果是0,直接CAS浅尝一下 // state 不是0,那就看下是不是锁重入操作 // 如果没抢到,或者不是锁重入操作,告辞,返回false public boolean tryLock() { // 非公平锁的竞争锁操作 return sync.nonfairTryAcquire(1); } final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }
-
tryLock(time,unit);
-
第一波分析,类似的代码:
// tryLock(time,unit)执行的方法 public final boolean tryAcquireNanos(int arg, long nanosTimeout)throws InterruptedException { // 线程的中断标记位,是不是从false,别改为了true,如果是,直接抛异常 if (Thread.interrupted()) throw new InterruptedException(); // tryAcquire分为公平和非公平锁两种执行方式,如果拿锁成功, 直接告辞, return tryAcquire(arg) || // 如果拿锁失败,在这要等待指定时间 doAcquireNanos(arg, nanosTimeout); } private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { // 如果等待时间是0秒,直接告辞,拿锁失败 if (nanosTimeout <= 0L) return false; // 设置结束时间。 final long deadline = System.nanoTime() + nanosTimeout; // 先扔到AQS队列 final Node node = addWaiter(Node.EXCLUSIVE); // 拿锁失败,默认true boolean failed = true; try { for (;;) { // 如果在AQS中,当前node是head的next,直接抢锁 final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return true; } // 结算剩余的可用时间 nanosTimeout = deadline - System.nanoTime(); // 判断是否是否用尽的位置 if (nanosTimeout <= 0L) return false; // shouldParkAfterFailedAcquire:根据上一个节点来确定现在是否可以挂起线程 if (shouldParkAfterFailedAcquire(p, node) && // 避免剩余时间太少,如果剩余时间少就不用挂起线程 nanosTimeout > spinForTimeoutThreshold) // 如果剩余时间足够,将线程挂起剩余时间 LockSupport.parkNanos(this, nanosTimeout); // 如果线程醒了,查看是中断唤醒的,还是时间到了唤醒的。 if (Thread.interrupted()) // 是中断唤醒的! throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
-
取消节点分析:
// 取消在AQS中排队的Node private void cancelAcquire(Node node) { // 如果当前节点为null,直接忽略。 if (node == null) return; //1. 线程设置为null node.thread = null; //2. 往前跳过被取消的节点,找到一个有效节点 Node pred = node.prev; while (pred.waitStatus > 0) node.prev = pred = pred.prev; //3. 拿到了上一个节点之前的next Node predNext = pred.next; //4. 当前节点状态设置为1,代表节点取消 node.waitStatus = Node.CANCELLED; // 脱离AQS队列的操作 // 当前Node是尾结点,将tail从当前节点替换为上一个节点 if (node == tail && compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { // 到这,上面的操作CAS操作失败 int ws = pred.waitStatus; // 不是head的后继节点 if (pred != head && // 拿到上一个节点的状态,只要上一个节点的状态不是取消状态,就改为-1 (ws == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) { // 上面的判断都是为了避免后面节点无法被唤醒。 // 前继节点是有效节点,可以唤醒后面的节点 Node next = node.next; if (next != null && next.waitStatus <= 0) compareAndSetNext(pred, predNext, next); } else { // 当前节点是head的后继节点 unparkSuccessor(node); } node.next = node; // help GC } }
-
lockInterruptibly方法
// 这个是lockInterruptibly和tryLock(time,unit)唯一的区别
// lockInterruptibly,拿不到锁资源,就死等,等到锁资源释放后,被唤醒,或者是被中断唤醒
private void doAcquireInterruptibly(int arg) throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
// 中断唤醒抛异常!
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
// 这个方法可以确认,当前挂起的线程,是被中断唤醒的,还是被正常唤醒的。
// 中断唤醒,返回true,如果是正常唤醒,返回false
return Thread.interrupted();
}
4、释放锁流程源码剖析
(1)释放锁流程概述
(2)释放锁源码分析
public void unlock() {
// 释放锁资源不分为公平锁和非公平锁,都是一个sync对象
sync.release(1);
}
// 释放锁的核心流程
public final boolean release(int arg) {
// 核心释放锁资源的操作之一
if (tryRelease(arg)) {
// 如果锁已经释放掉了,走这个逻辑
Node h = head;
// h不为null,说明有排队的(录课时估计脑袋蒙圈圈。)
// 如果h的状态不为0(为-1),说明后面有排队的Node,并且线程已经挂起了。
if (h != null && h.waitStatus != 0)
// 唤醒排队的线程
unparkSuccessor(h);
return true;
}
return false;
}
// ReentrantLock释放锁资源操作
protected final boolean tryRelease(int releases) {
// 拿到state - 1(并没有赋值给state)
int c = getState() - releases;
// 判断当前持有锁的线程是否是当前线程,如果不是,直接抛出异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
// free,代表当前锁资源是否释放干净了。
boolean free = false;
if (c == 0) {
// 如果state - 1后的值为0,代表释放干净了。
free = true;
// 将持有锁的线程置位null
setExclusiveOwnerThread(null);
}
// 将c设置给state
setState(c);
// 锁资源释放干净返回true,否则返回false
return free;
}
// 唤醒后面排队的Node
private void unparkSuccessor(Node node) {
// 拿到头节点状态
int ws = node.waitStatus;
if (ws < 0)
// 先基于CAS,将节点状态从-1,改为0
compareAndSetWaitStatus(node, ws, 0);
// 拿到头节点的后续节点。
Node s = node.next;
// 如果后续节点为null或者,后续节点的状态为1,代表节点取消了。
if (s == null || s.waitStatus > 0) {
s = null;
// 如果后续节点为null,或者后续节点状态为取消状态,从后往前找到一个有效节点环境
for (Node t = tail; t != null && t != node; t = t.prev)
// 从后往前找到状态小于等于0的节点
// 找到离head最新的有效节点,并赋值给s
if (t.waitStatus <= 0)
s = t;
}
// 只要找到了这个需要被唤醒的节点,执行unpark唤醒
if (s != null)
LockSupport.unpark(s.thread);
}
5、AQS常见的问题
(1)AQS中为什么要有一个虚拟的head节点
因为AQS提供了ReentrantLock的基本实现,而在ReentrantLock释放锁资源时,需要去考虑是否需要执行unparkSuccessor方法,去唤醒后继节点。
因为Node中存在waitStatus的状态,默认情况下状态为0,如果当前节点的后继节点线程挂起了,那么就将当前节点的状态设置为-1。这个-1状态的出现是为了避免重复唤醒或者释放资源的问题。
因为AQS中排队的Node中的线程如果挂起了,是无法自动唤醒的。需要释放锁或者释放资源后,再被释放的线程去唤醒挂起的线程。 因为唤醒节点需要从整个AQS双向链表中找到离head最近的有效节点去唤醒。而这个找离head最近的Node可能需要遍历整个双向链表。如果AQS中,没有挂起的线程,代表不需要去遍历AQS双向链表去找离head最近的有效节点。
为了避免出现不必要的循环链表操作,提供了一个-1的状态。如果只有一个Node进入到AQS中排队,所以发现如果是第一个Node进来,他必须先初始化一个虚拟的head节点作为头,来监控后继节点中是否有挂起的线程。
(2)AQS中为什么选择使用双向链表,而不是单向链表
首先AQS中一般是存放没有获取到资源的Node,而在竞争锁资源时,ReentrantLock提供了一个方法,lockInterruptibly方法,也就是线程在竞争锁资源的排队途中,允许中断。中断后会执行cancelAcquire方法,从而将当前节点状态置位1,并且从AQS队列中移除掉。如果采用单向链表,当前节点只能按到后继或者前继节点,这样是无法将前继节点指向后继节点的,需要遍历整个AQS从头或者从尾去找。单向链表在移除AQS中排队的Node时,成本很高。
当前在唤醒后继节点时,如果是单向链表也会出问题,因为节点插入方式的问题,导致只能单向的去找有效节点去唤醒,从而造成很多次无效的遍历操作,如果是双向链表就可以解决这个问题。
6、ConditionObject
(1)ConditionObject的介绍&应用
像synchronized提供了wait和notify的方法实现线程在持有锁时,可以实现挂起,已经唤醒的操作。
ReentrantLock也拥有这个功能。
ReentrantLock提供了await和signal方法去实现类似wait和notify的功能。
想执行await或者是signal就必须先持有lock锁的资源。
先look一下Condition的应用
public static void main(String[] args) throws InterruptedException, IOException {
ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();
new Thread(() -> {
lock.lock();
System.out.println("子线程获取锁资源并await挂起线程");
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("子线程挂起后被唤醒!持有锁资源");
}).start();
Thread.sleep(100);
// =================main======================
lock.lock();
System.out.println("主线程等待5s拿到锁资源,子线程执行了await方法");
condition.signal();
System.out.println("主线程唤醒了await挂起的子线程");
lock.unlock();
}
(2)Condition的构建方式&核心属性
发现在通过lock锁对象执行newCondition方法时,本质就是直接new的AQS提供的ConditionObject对象
final ConditionObject newCondition() {
return new ConditionObject();
}
其实lock锁中可以有多个Condition对象。
在对Condition1进行操作时,不会影响到Condition2的单向链表。
其次可以发现ConditionObject中,只有两个核心属性:
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
虽然Node对象有prev和next,但是在ConditionObject中是不会使用这两个属性的,只要在Condition队列中,这两个属性都是null。在ConditionObject中只会使用nextWaiter的属性实现单向链表的效果。
(3)Condition的await方法分析(前置分析)
持有锁的线程在执行await方法后会做几个操作:
- 判断线程是否中断,如果中断了,什么都不做。
- 没有中断,就讲当前线程封装为Node添加到Condition的单向链表中
- 一次性释放掉锁资源。
- 如果当前线程没有在AQS队列,就正常执行LockSupport.park(this)挂起线程。
// await方法的前置分析,只分析到线程挂起
public final void await() throws InterruptedException {
// 先判断线程的中断标记位是否是true
if (Thread.interrupted())
// 如果是true,就没必要执行后续操作挂起了。
throw new InterruptedException();
// 在线程挂起之前,先将当前线程封装为Node,并且添加到Condition队列中
Node node = addConditionWaiter();
// fullyRelease在释放锁资源,一次性将锁资源全部释放,并且保留重入的次数
int savedState = fullyRelease(node);
// 省略一行代码……
// 当前Node是否在AQS队列中?
// 执行fullyRelease方法后,线程就释放锁资源了,如果线程刚刚释放锁资源,其他线程就立即执行了signal方法,
// 此时当前线程就被放到了AQS的队列中,这样一来线程就不需要执行LockSupport.park(this);去挂起线程了
while (!isOnSyncQueue(node)) {
// 如果没有在AQS队列中,正常在Condition单向链表里,正常挂起线程。
LockSupport.park(this);
// 省略部分代码……
}
// 省略部分代码……
}
// 线程挂起先,添加到Condition单向链表的业务~~
private Node addConditionWaiter() {
// 拿到尾节点。
Node t = lastWaiter;
// 如果尾节点有值,并且尾节点的状态不正常,不是-2,尾节点可能要拜拜了~
if (t != null && t.waitStatus != Node.CONDITION) {
// 如果尾节点已经取消了,需要干掉取消的尾节点~
unlinkCancelledWaiters();
// 重新获取lastWaiter
t = lastWaiter;
}
// 构建当前线程的Node,并且状态设置为-2
Node node = new Node(Thread.currentThread(), Node.CONDITION);
// 如果last节点为null。直接将当前节点设置为firstWaiter
if (t == null)
firstWaiter = node;
else
// 如果last节点不为null,说明有值,就排在lastWaiter的后面
t.nextWaiter = node;
// 把当前节点设置为最后一个节点
lastWaiter = node;
// 返回当前节点
return node;
}
// 干掉取消的尾节点。
private void unlinkCancelledWaiters() {
// 拿到头节点
Node t = firstWaiter;
// 声明一个节点,爱啥啥~~~
Node trail = null;
// 如果t不为null,就正常执行~~
while (t != null) {
// 拿到t的next节点
Node next = t.nextWaiter;
// 如果t的状态不为-2,说明有问题
if (t.waitStatus != Node.CONDITION) {
// t节点的next为null
t.nextWaiter = null;
// 如果trail为null,代表头结点状态就是1,
if (trail == null)
// 将头结点指向next节点
firstWaiter = next;
else
// 如果trail有值,说明不是头结点位置
trail.nextWaiter = next;
// 如果next为null,说明单向链表遍历到最后了,直接结束
if (next == null)
lastWaiter = trail;
}
// 如果t的状态是-2,一切正常
else {
// 临时存储t
trail = t;
}
// t指向之前的next
t = next;
}
}
// 一次性释放锁资源
final int fullyRelease(Node node) {
// 标记位,释放锁资源默认失败!
boolean failed = true;
try {
// 拿到现在state的值
int savedState = getState();
// 一次性释放干净全部锁资源
if (release(savedState)) {
// 释放锁资源失败了么? 没有!
failed = false;
// 返回对应的锁资源信息
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
// 如果释放锁资源失败,将节点状态设置为取消
node.waitStatus = Node.CANCELLED;
}
}
(4)Condition的signal方法分析
分为了几个部分:
- 确保执行signal方法的是持有锁的线程
- 脱离Condition的队列
- 将Node状态从-2改为0
- 将Node添加到AQS队列
- 为了避免当前Node无法在AQS队列正常唤醒做了一些判断和操作
// 线程挂起后,可以基于signal唤醒~
public final void signal() {
// 在ReentrantLock中,如果执行signal的线程没有持有锁资源,直接扔异常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 拿到排在Condition首位的Node
Node first = firstWaiter;
// 有Node在排队,才需要唤醒,如果没有,直接告辞~~
if (first != null)
doSignal(first);
}
// 开始唤醒Condition中的Node中的线程
private void doSignal(Node first) {
// 先一波do-while走你~~~
do {
// 获取到第二个节点,并且将第二个节点设置为firstWaiter
if ( (firstWaiter = first.nextWaiter) == null)
// 说明就一个节点在Condition队列中,那么直接将firstWaiter和lastWaiter置位null
lastWaiter = null;
// 如果还有nextWaiter节点,因为当前节点要被唤醒了,脱离整个Condition队列。将nextWaiter置位null
first.nextWaiter = null;
// 如果transferForSignal返回true,一切正常,退出while循环
} while (!transferForSignal(first) &&
// 如果后续节点还有,往后面继续唤醒,如果没有,退出while循环
(first = firstWaiter) != null);
}
// 准备开始唤醒在Condition中排队的Node
final boolean transferForSignal(Node node) {
// 将在Condition队列中的Node的状态从-2,改为0,代表要扔到AQS队列了。
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
// 如果失败了,说明在signal之前应当是线程被中断了,从而被唤醒了。
return false;
// 如果正常的将Node的状态从-2改为0,这是就要将Condition中的这个Node扔到AQS的队列。
// 将当前Node扔到AQS队列,返回的p是当前Node的prev
Node p = enq(node);
// 获取上一个Node的状态
int ws = p.waitStatus;
// 如果ws > 0 ,说明这个Node已经被取消了。
// 如果ws状态不是取消,将prev节点的状态改为-1,。
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
// 如果prev节点已经取消了,可能会导致当前节点永远无法被唤醒。立即唤醒当前节点,基于acquireQueued方法,
// 让当前节点找到一个正常的prev节点,并挂起线程
// 如果prev节点正常,但是CAS修改prev节点失败了。证明prev节点因为并发原因导致状态改变。还是为了避免当前
// 节点无法被正常唤醒,提前唤醒当前线程,基于acquireQueued方法,让当前节点找到一个正常的prev节点,并挂起线程
LockSupport.unpark(node.thread);
// 返回true
return true;
}
(5)Conditiond的await方法分析(后置分析)
分为了几个部分:
- 唤醒之后,要先确认是中断唤醒还是signal唤醒,还是signal唤醒后被中断
- 确保当前线程的Node已经在AQS队列中
- 执行acquireQueued方法,等待锁资源。
- 在获取锁资源后,要确认是否在获取锁资源的阶段被中断过,如果被中断过,并且不是THROW_IE,那就确保interruptMode是REINTERRUPT。
- 确认当前Node已经不在Condition队列中了
- 最终根据interruptMode来决定具体做的事情
- 0:嘛也不做。
- THROW_IE:抛出异常
- REINTERRUPT:执行线程的interrupt方法
// 现在分析await方法的后半部分
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
// 中断模式~
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
// 如果线程执行到这,说明现在被唤醒了。
// 线程可以被signal唤醒。(如果是signal唤醒,可以确认线程已经在AQS队列中)
// 线程可以被interrupt唤醒,线程被唤醒后,没有在AQS队列中。
// 如果线程先被signal唤醒,然后线程中断了。。。。(做一些额外处理)
// checkInterruptWhileWaiting可以确认当前中如何唤醒的。
// 返回的值,有三种
// 0:正常signal唤醒,没别的事(不知道Node是否在AQS队列)
// THROW_IE(-1):中断唤醒,并且可以确保在AQS队列
// REINTERRUPT(1):signal唤醒,但是线程被中断了,并且可以确保在AQS队列
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// Node一定在AQS队列
// 执行acquireQueued,尝试在ReentrantLock中获取锁资源。
// acquireQueued方法返回true:代表线程在AQS队列中挂起时,被中断过
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
// 如果线程在AQS队列排队时,被中断了,并且不是THROW_IE状态,确保线程的interruptMode是REINTERRUPT
// REINTERRUPT:await不是中断唤醒,但是后续被中断过!!!
interruptMode = REINTERRUPT;
// 如果当前Node还在condition的单向链表中,脱离Condition的单向链表
if (node.nextWaiter != null)
unlinkCancelledWaiters();
// 如果interruptMode是0,说明线程在signal后以及持有锁的过程中,没被中断过,什么事都不做!
if (interruptMode != 0)
// 如果不是0~
reportInterruptAfterWait(interruptMode);
}
// 判断当前线程被唤醒的模式,确认interruptMode的值。
private int checkInterruptWhileWaiting(Node node) {
// 判断线程是否中断了。
return Thread.interrupted() ?
// THROW_IE:代表线程是被interrupt唤醒的,需要向上排除异常
// REINTERRUPT:代表线程是signal唤醒的,但是在唤醒之后,被中断了。
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
// 线程是正常的被signal唤醒,并且线程没有中断过。
0;
}
// 判断线程到底是中断唤醒的,还是signal唤醒的!
final boolean transferAfterCancelledWait(Node node) {
// 基于CAS将Node的状态从-2改为0
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
// 说明是中断唤醒的线程。因为CAS成功了。
// 将Node添加到AQS队列中~(如果是中断唤醒的,当前线程同时存在Condition的单向链表以及AQS的队列中)
enq(node);
// 返回true
return true;
}
// 判断当前的Node是否在AQS队列(signal唤醒的,但是可能线程还没放到AQS队列)
// 等到signal方法将线程的Node扔到AQS队列后,再做后续操作
while (!isOnSyncQueue(node))
// 如果没在AQS队列上,那就线程让步,稍等一会,Node放到AQS队列再处理(看CPU)
Thread.yield();
// signal唤醒的,返回false
return false;
}
// 确认Node是否在AQS队列上
final boolean isOnSyncQueue(Node node) {
// 如果线程状态为-2,肯定没在AQS队列
// 如果prev节点的值为null,肯定没在AQS队列
if (node.waitStatus == Node.CONDITION || node.prev == null)
// 返回false
return false;
// 如果节点的next不为null。说明已经在AQS队列上。、
if (node.next != null)
// 确定AQS队列上有!
return true;
// 如果上述判断都没有确认节点在AQS队列上,在AQS队列中寻找一波
return findNodeFromTail(node);
}
// 在AQS队列中找当前节点
private boolean findNodeFromTail(Node node) {
// 拿到尾节点
Node t = tail;
for (;;) {
// tail是否是当前节点,如果是,说明在AQS队列
if (t == node)
// 可以跳出while循环
return true;
// 如果节点为null,AQS队列中没有当前节点
if (t == null)
// 进入while,让步一手
return false;
// t向前引用
t = t.prev;
}
}
private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {
// 如果是中断唤醒的await,直接抛出异常!
if (interruptMode == THROW_IE)
throw new InterruptedException();
// 如果是REINTERRUPT,signal后被中断过
else if (interruptMode == REINTERRUPT)
// 确认线程的中断标记位是true
// Thread.currentThread().interrupt();
selfInterrupt();
}
(6)Condition的awaitNanos&signalAll方法分析
awaitNanos:仅仅是在await方法的基础上,做了一内内的改变,整体的逻辑思想都是一样的。
挂起线程时,传入要阻塞的时间,时间到了,自动唤醒,走添加到AQS队列的逻辑
// await指定时间,多了个时间到了自动醒。
public final long awaitNanos(long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
// deadline:当前线程最多挂起到什么时间点
final long deadline = System.nanoTime() + nanosTimeout;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
// nanosTimeout的时间小于等于0,直接告辞!!
if (nanosTimeout <= 0L) {
// 正常扔到AQS队列
transferAfterCancelledWait(node);
break;
}
// nanosTimeout的时间大于1000纳秒时,才可以挂起线程
if (nanosTimeout >= spinForTimeoutThreshold)
// 如果大于,正常挂起
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
// 计算剩余的挂起时间,可能需要重新的走while循环,再次挂起线程
nanosTimeout = deadline - System.nanoTime();
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
// 剩余的挂起时间
return deadline - System.nanoTime();
}
signalAll方法。这个方法一看就懂,之前signal是唤醒1个,这个是全部唤醒
// 以do-while的形式,将Condition单向链表中的所有Node,全部唤醒并扔到AQS队列
private void doSignalAll(Node first) {
// 将头尾都置位null~
lastWaiter = firstWaiter = null;
do {
// 拿到next节点的引用
Node next = first.nextWaiter;
// 断开当前Node的nextWaiter
first.nextWaiter = null;
// 修改Node状态,扔AQS队列,是否唤醒!
transferForSignal(first);
// 指向下一个节点
first = next;
} while (first != null);
}
四、深入ReentrantReadWriteLock
1、为什么要出现读写锁
synchronized和ReentrantLock都是互斥锁。
如果说有一个操作是读多写少的,还要保证线程安全的话。如果采用上述的两种互斥锁,效率方面很定是很低的。
在这种情况下,咱们就可以使用ReentrantReadWriteLock读写锁去实现。
读读之间是不互斥的,可以读和读操作并发执行。
但是如果涉及到了写操作,那么还得是互斥的操作。
static ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
static ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
static ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
readLock.lock();
try {
System.out.println("子线程!");
try {
Thread.sleep(500000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} finally {
readLock.unlock();
}
}).start();
Thread.sleep(1000);
writeLock.lock();
try {
System.out.println("主线程!");
} finally {
writeLock.unlock();
}
}
2、读写锁的实现原理
ReentrantReadWriteLock还是基于AQS实现的,还是对state进行操作,拿到锁资源就去干活,如果没有拿到,依然去AQS队列中排队。
读锁操作:基于state的高16位进行操作。
写锁操作:基于state的低16为进行操作。
ReentrantReadWriteLock依然是可重入锁。
写锁重入:读写锁中的写锁的重入方式,基本和ReentrantLock一致,没有什么区别,依然是对state进行+1操作即可,只要确认持有锁资源的线程,是当前写锁线程即可。只不过之前ReentrantLock的重入次数是state的正数取值范围,但是读写锁中写锁范围就变小了。
读锁重入:因为读锁是共享锁。读锁在获取锁资源操作时,是要对state的高16位进行 + 1操作。因为读锁是共享锁,所以同一时间会有多个读线程持有读锁资源。这样一来,多个读操作在持有读锁时,无法确认每个线程读锁重入的次数。为了去记录读锁重入的次数,每个读操作的线程,都会有一个ThreadLocal记录锁重入的次数。
写锁的饥饿问题:读锁是共享锁,当有线程持有读锁资源时,再来一个线程想要获取读锁,直接对state修改即可。在读锁资源先被占用后,来了一个写锁资源,此时,大量的需要获取读锁的线程来请求锁资源,如果可以绕过写锁,直接拿资源,会造成写锁长时间无法获取到写锁资源。
读锁在拿到锁资源后,如果再有读线程需要获取读锁资源,需要去AQS队列排队。如果队列的前面需要写锁资源的线程,那么后续读线程是无法拿到锁资源的。持有读锁的线程,只会让写锁线程之前的读线程拿到锁资源
3、写锁分析
(1)写锁加锁流程概述
(2)写锁加锁源码分析
写锁加锁流程
// 写锁加锁的入口
public void lock() {
sync.acquire(1);
}
// 阿巴阿巴!!
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
// 读写锁的写锁实现tryAcquire
protected final boolean tryAcquire(int acquires) {
// 拿到当前线程
Thread current = Thread.currentThread();
// 拿到state的值
int c = getState();
// 得到state低16位的值
int w = exclusiveCount(c);
// 判断是否有线程持有着锁资源
if (c != 0) {
// 当前没有线程持有写锁,读写互斥,告辞。
// 有线程持有写锁,持有写锁的线程不是当前线程,不是锁重入,告辞。
if (w == 0 || current != getExclusiveOwnerThread())
return false;
// 当前线程持有写锁。 锁重入。
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 没有超过锁重入的次数,正常 + 1
setState(c + acquires);
return true;
}
// 尝试获取锁资源
if (writerShouldBlock() ||
// CAS拿锁
!compareAndSetState(c, c + acquires))
return false;
// 拿锁成功,设置占有互斥锁的线程
setExclusiveOwnerThread(current);
// 返回true
return true;
}
// ================================================================
// 这个方法是将state的低16位的值拿到
int w = exclusiveCount(c);
state & ((1 << 16) - 1)
00000000 00000000 00000000 00000001 == 1
00000000 00000001 00000000 00000000 == 1 << 16
00000000 00000000 11111111 11111111 == (1 << 16) - 1
&运算,一个为0,必然为0,都为1,才为1
// ================================================================
// writerShouldBlock方法查看公平锁和非公平锁的效果
// 非公平锁直接返回false执行CAS尝试获取锁资源
// 公平锁需要查看是否有排队的,如果有排队的,我是否是head的next
(3)写锁释放锁流程概述&释放锁源码
释放的流程和ReentrantLock一致,只是在判断释放是否干净时,判断低16位的值
// 写锁释放锁的tryRelease方法
protected final boolean tryRelease(int releases) {
// 判断当前持有写锁的线程是否是当前线程
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 获取state - 1
int nextc = getState() - releases;
// 判断低16位结果是否为0,如果为0,free设置为true
boolean free = exclusiveCount(nextc) == 0;
if (free)
// 将持有锁的线程设置为null
setExclusiveOwnerThread(null);
// 设置给state
setState(nextc);
// 释放干净,返回true。 写锁有冲入,这里需要返回false,不去释放排队的Node
return free;
}
4、读锁分析
(1)读锁加锁流程概述
1、分析读锁加速的基本流程
2、分析读锁的可重入锁实现以及优化
3、解决ThreadLocal内存泄漏问题
4、读锁获取锁自后,如果唤醒AQS中排队的读线程
(2)基础读锁流程
针对上述简单逻辑的源码分析
// 读锁加锁的方法入口
public final void acquireShared(int arg) {
// 竞争锁资源滴干活
if (tryAcquireShared(arg) < 0)
// 没拿到锁资源,去排队
doAcquireShared(arg);
}
// 读锁竞争锁资源的操作
protected final int tryAcquireShared(int unused) {
// 拿到当前线程
Thread current = Thread.currentThread();
// 拿到state
int c = getState();
// 拿到state的低16位,判断 != 0,有写锁占用着锁资源
// 并且,当前占用锁资源的线程不是当前线程
if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
// 写锁被其他线程占用,无法获取读锁,直接返回 -1,去排队
return -1;
// 没有线程持有写锁、当前线程持有写锁
// 获取读锁的信息,state的高16位。
int r = sharedCount(c);
// 公平锁:就查看队列是由有排队的,有排队的,直接告辞,进不去if,后面也不用判断(没人排队继续走)
// 非公平锁:没有排队的,直接抢。 有排队的,但是读锁其实不需要排队,如果出现这个情况,大部分是写锁资源刚刚释放,
// 后续Node还没有来记得拿到读锁资源,当前竞争的读线程,可以直接获取
if (!readerShouldBlock() &&
// 判断持有读锁的临界值是否达到
r < MAX_COUNT &&
// CAS修改state,对高16位进行 + 1
compareAndSetState(c, c + SHARED_UNIT)) {
// 省略部分代码!!!!
return 1;
}
return fullTryAcquireShared(current);
}
// 非公平锁的判断
final boolean apparentlyFirstQueuedIsExclusive() {
Node h, s;
return (h = head) != null && // head为null,可以直接抢占锁资源
(s = h.next) != null && // head的next为null,可以直接抢占锁资源
!s.isShared() && // 如果排在head后面的Node,是共享锁,可以直接抢占锁资源。
s.thread != null; // 后面排队的thread为null,可以直接抢占锁资源
}
(3)读锁重入流程
=============重入操作
前面阐述过,读锁为了记录锁重入的次数,需要让每个读线程用ThreadLocal存储重入次数
ReentrantReadWriteLock对读锁重入做了一些优化操作
============记录重入次数的核心
ReentrantReadWriteLock在内部对ThreadLocal做了封装,基于HoldCount的对象存储重入次数,在内部有个count属性记录,而且每个线程都是自己的ThreadLocalHoldCounter,所以可以直接对内部的count进行++操作。
=============第一个获取读锁资源的重入次数记录方式
第一个拿到读锁资源的线程,不需要通过ThreadLocal存储,内部提供了两个属性来记录第一个拿到读锁资源线程的信息
内部提供了firstReader记录第一个拿到读锁资源的线程,firstReaderHoldCount记录firstReader的锁重入次数
==============最后一个获取读锁资源的重入次数记录方式
最后一个拿到读锁资源的线程,也会缓存他的重入次数,这样++起来更方便
基于cachedHoldCounter缓存最后一个拿到锁资源现成的重入次数
==============最后一个获取读锁资源的重入次数记录方式
重入次数的流程执行方式:
1、判断当前线程是否是第一个拿到读锁资源的:如果是,直接将firstReader以及firstReaderHoldCount设置为当前线程的信息
2、判断当前线程是否是firstReader:如果是,直接对firstReaderHoldCount++即可。
3、跟firstReader没关系了,先获取cachedHoldCounter,判断是否是当前线程。
3.1、如果不是,获取当前线程的重入次数,将cachedHoldCounter设置为当前线程。
3.2、如果是,判断当前重入次数是否为0,重新设置当前线程的锁从入信息到readHolds(ThreadLocal)中,算是初始化操作,重入次数是0
3.3、前面两者最后都做count++
上述逻辑源码分析
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
// ===============================================================
// 判断r == 0,当前是第一个拿到读锁资源的线程
if (r == 0) {
// 将firstReader设置为当前线程
firstReader = current;
// 将count设置为1
firstReaderHoldCount = 1;
}
// 判断当前线程是否是第一个获取读锁资源的线程
else if (firstReader == current) {
// 直接++。
firstReaderHoldCount++;
}
// 到这,就说明不是第一个获取读锁资源的线程
else {
// 那获取最后一个拿到读锁资源的线程
HoldCounter rh = cachedHoldCounter;
// 判断当前线程是否是最后一个拿到读锁资源的线程
if (rh == null || rh.tid != getThreadId(current))
// 如果不是,设置当前线程为cachedHoldCounter
cachedHoldCounter = rh = readHolds.get();
// 当前线程是之前的cacheHoldCounter
else if (rh.count == 0)
// 将当前的重入信息设置到ThreadLocal中
readHolds.set(rh);
// 重入的++
rh.count++;
}
// ===============================================================
return 1;
}
return fullTryAcquireShared(current);
}
(4)读锁加锁的后续逻辑fullTryAcquireShared
// tryAcquireShard方法中,如果没有拿到锁资源,走这个方法,尝试再次获取,逻辑跟上面基本一致。
final int fullTryAcquireShared(Thread current) {
// 声明当前线程的锁重入次数
HoldCounter rh = null;
// 死循环
for (;;) {
// 再次拿到state
int c = getState();
// 当前如果有写锁在占用锁资源,并且不是当前线程,返回-1,走排队策略
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)
return -1;
}
// 查看当前是否可以尝试竞争锁资源(公平锁和非公平锁的逻辑)
else if (readerShouldBlock()) {
// 无论公平还是非公平,只要进来,就代表要放到AQS队列中了,先做一波准备
// 在处理ThreadLocal的内存泄漏问题
if (firstReader == current) {
// 如果当前当前线程是之前的firstReader,什么都不用做
} else {
// 第一次进来是null。
if (rh == null) {
// 拿到最后一个获取读锁的线程
rh = cachedHoldCounter;
// 当前线程并不是cachedHoldCounter,没到拿到
if (rh == null || rh.tid != getThreadId(current)) {
// 从自己的ThreadLocal中拿到重入计数器
rh = readHolds.get();
// 如果计数器为0,说明之前没拿到过读锁资源
if (rh.count == 0)
// remove,避免内存泄漏
readHolds.remove();
}
}
// 前面处理完之后,直接返回-1
if (rh.count == 0)
return -1;
}
}
// 判断重入次数,是否超出阈值
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// CAS尝试获取锁资源
if (compareAndSetState(c, c + SHARED_UNIT)) {
if (sharedCount(c) == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
if (rh == null)
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
cachedHoldCounter = rh; // cache for release
}
return 1;
}
}
}
(5)读线程在AQS队列获取锁资源的后续操作
1、正常如果都是读线程来获取读锁资源,不需要使用到AQS队列的,直接CAS操作即可
2、如果写线程持有着写锁,这是读线程就需要进入到AQS队列排队,可能会有多个读线程在AQS中。
当写锁释放资源后,会唤醒head后面的读线程,当head后面的读线程拿到锁资源后,还需要查看next节点是否也是读线程在阻塞,如果是,直接唤醒
源码分析
// 读锁需要排队的操作
private void doAcquireShared(int arg) {
// 声明Node,类型是共享锁,并且扔到AQS中排队
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 拿到上一个节点
final Node p = node.predecessor();
// 如果prev节点是head,直接可以执行tryAcquireShared
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
// 拿到读锁资源后,需要做的后续处理
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
// 找到prev有效节点,将状态设置为-1,挂起当前线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private void setHeadAndPropagate(Node node, int propagate) {
// 拿到head节点
Node h = head;
// 将当前节点设置为head节点
setHead(node);
// 第一个判断更多的是在信号量有处理JDK1.5 BUG的操作。
if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
// 拿到当前Node的next节点
Node s = node.next;
// 如果next节点是共享锁,直接唤醒next节点
if (s == null || s.isShared())
doReleaseShared();
}
}
(6)读锁的释放锁流程
1、处理重入以及state的值
2、唤醒后续排队的Node
源码分析
// 读锁释放锁流程
public final boolean releaseShared(int arg) {
// tryReleaseShared:处理state的值,以及可重入的内容
if (tryReleaseShared(arg)) {
// AQS队列的事!
doReleaseShared();
return true;
}
return false;
}
// 1、 处理重入问题 2、 处理state
protected final boolean tryReleaseShared(int unused) {
// 拿到当前线程
Thread current = Thread.currentThread();
// 如果是firstReader,直接干活,不需要ThreadLocal
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
}
// 不是firstReader,从cachedHoldCounter以及ThreadLocal处理
else {
// 如果是cachedHoldCounter,正常--
HoldCounter rh = cachedHoldCounter;
// 如果不是cachedHoldCounter,从自己的ThreadLocal中拿
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
int count = rh.count;
// 如果为1或者更小,当前线程就释放干净了,直接remove,避免value内存泄漏
if (count <= 1) {
readHolds.remove();
// 如果已经是0,没必要再unlock,扔个异常
if (count <= 0)
throw unmatchedUnlockException();
}
// -- 走你。
--rh.count;
}
for (;;) {
// 拿到state,高16位,-1,成功后,返回state是否为0
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
// 唤醒AQS中排队的线程
private void doReleaseShared() {
// 死循环
for (;;) {
// 拿到头
Node h = head;
// 说明有排队的
if (h != null && h != tail) {
// 拿到head的状态
int ws = h.waitStatus;
// 判断是否为 -1
if (ws == Node.SIGNAL) {
// 到这,说明后面有挂起的线程,先基于CAS将head的状态从-1,改为0
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 唤醒后续节点
unparkSuccessor(h);
}
// 这里不是给读写锁准备的,在信号量里说。。。
else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
// 这里是出口
if (h == head)
break;
}
}
第四章、JUC并发工具
一、CountDownLatch应用&源码分析
1、CountDownLatch介绍
CountDownLatch就是JUC包下的一个工具,整个工具最核心的功能就是计数器。
如果有三个业务需要并行处理,并且需要知道三个业务全部都处理完毕了。
需要一个并发安全的计数器来操作。
CountDownLatch就可以实现。
给CountDownLatch设置一个数值。可以设置3。
每个业务处理完毕之后,执行一次countDown方法,指定的3每次在执行countDown方法时,对3进行-1。
主线程可以在业务处理时,执行await,主线程会阻塞等待任务处理完毕。
当设置的3基于countDown方法减为0之后,主线程就会被唤醒,继续处理后续业务。
当咱们的业务中,出现2个以上允许并行处理的任务,并且需要在任务都处理完毕后,再做其他处理时,可以采用CountDownLatch去实现这个功能。
2、CountDownLatch应用
模拟有三个任务需要并行处理,在三个任务全部处理完毕后,再执行后续操作
CountDownLatch中,执行countDown方法,代表一个任务结束,对计数器 - 1
执行await方法,代表等待计数器变为0时,再继续执行
执行await(time,unit)方法,代表等待time时长,如果计数器不为0,返回false,如果在等待期间,计数器为0,方法就返回true
一般CountDownLatch更多的是基于业务去构建,不采用成员变量。
static ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(3);
static CountDownLatch countDownLatch = new CountDownLatch(3);
public static void main(String[] args) throws InterruptedException {
System.out.println("主业务开始执行");
sleep(1000);
executor.execute(CompanyTest::a);
executor.execute(CompanyTest::b);
executor.execute(CompanyTest::c);
System.out.println("三个任务并行执行,主业务线程等待");
// 死等任务结束
// countDownLatch.await();
// 如果在规定时间内,任务没有结束,返回false
if (countDownLatch.await(10, TimeUnit.SECONDS)) {
System.out.println("三个任务处理完毕,主业务线程继续执行");
}else{
System.out.println("三个任务没有全部处理完毕,执行其他的操作");
}
}
private static void a() {
System.out.println("A任务开始");
sleep(1000);
System.out.println("A任务结束");
countDownLatch.countDown();
}
private static void b() {
System.out.println("B任务开始");
sleep(1500);
System.out.println("B任务结束");
countDownLatch.countDown();
}
private static void c() {
System.out.println("C任务开始");
sleep(2000);
System.out.println("C任务结束");
countDownLatch.countDown();
}
private static void sleep(long timeout){
try {
Thread.sleep(timeout);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
3、CountDownLatch源码分析
保证CountDownLatch就是一个计数器,没有什么特殊的功能,查看源码也只是查看计数器实现的方式
发现CountDownLatch的内部类Sync继承了AQS,CountDownLatch就是基于AQS实现的计数器。
AQS就是一个state属性,以及AQS双向链表
猜测计数器的数值实现就是基于state去玩的。
主线程阻塞的方式,也是阻塞在了AQS双向链表中。
(1)有参构造
就是构建内部类Sync,并且给AQS中的state赋值
// CountDownLatch的有参构造
public CountDownLatch(int count) {
// 健壮性校验
if (count < 0) throw new IllegalArgumentException("count < 0");
// 构建内部类,Sync传入count
this.sync = new Sync(count);
}
// AQS子类,Sync的有参构造
Sync(int count) {
// 就是给AQS中的state赋值
setState(count);
}
(2)await方法
await方法就时判断当前CountDownLatch中的state是否为0,如果为0,直接正常执行后续任务
如果不为0,以共享锁的方式,插入到AQS的双向链表,并且挂起线程
// 一般主线程await的方法,阻塞主线程,等待state为0
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
// 执行了AQS的acquireSharedInterruptibly方法
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
// 判断线程是否中断,如果中断标记位是true,直接抛出异常
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
// 共享锁挂起的操作
doAcquireSharedInterruptibly(arg);
}
// tryAcquireShared在CountDownLatch中的实现
protected int tryAcquireShared(int acquires) {
// 查看state是否为0,如果为0,返回1,不为0,返回-1
return (getState() == 0) ? 1 : -1;
}
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
// 封装当前先成为Node,属性为共享锁
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 在这,就需要挂起当前线程。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
(3)countDown方法
countDown方法本质就是对state - 1,如果state - 1后变为0,需要去AQS的链表中唤醒挂起的节点
// countDown对计数器-1
public void countDown() {
// 是-1。
sync.releaseShared(1);
}
// AQS提供的功能
public final boolean releaseShared(int arg) {
// 对state - 1
if (tryReleaseShared(arg)) {
// state - 1后,变为0,执行doReleaseShared
doReleaseShared();
return true;
}
return false;
}
// CountDownLatch的tryReleaseShared实现
protected boolean tryReleaseShared(int releases) {
// 死循环是为了避免CAS并发问题
for (;;) {
// 获取state
int c = getState();
// state已经为0,直接返回false
if (c == 0)
return false;
// 对获取到的state - 1
int nextc = c-1;
// 基于CAS的方式,将值赋值给state
if (compareAndSetState(c, nextc))
// 赋值完,发现state为0了。此时可能会有线程在await方法处挂起,那边挂起,需要这边唤醒
return nextc == 0;
}
}
// 如何唤醒在await方法处挂起的线程
private void doReleaseShared() {
// 死循环
for (;;) {
// 拿到head
Node h = head;
// head不为null,有值,并且head != tail,代表至少2个节点
// 一个虚拟的head,加上一个实质性的Node
if (h != null && h != tail) {
// 说明AQS队列中有节点
int ws = h.waitStatus;
// 如果head节点的状态为 -1.
if (ws == Node.SIGNAL) {
// 先对head节点将状态从-1,修改为0,避免重复唤醒的情况
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 正常唤醒节点即可,先看head.next,能唤醒就唤醒,如果head.next有问题,从后往前找有效节点
unparkSuccessor(h);
}
// 会在Semaphore中谈到这个位置
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
// 会在Semaphore中谈到这个位置
if (h == head)
break;
}
}
二、CyclicBarrier应用&源码分析
1、CyclicBarrier介绍
从名字上来看CyclicBarrier,就是代表循环屏障
Barrier屏障:让一个或多个线程达到一个屏障点,会被阻塞。屏障点会有一个数值,当达到一个线程阻塞在屏障点时,就会对屏障点的数值进行-1操作,当屏障点数值减为0时,屏障就会打开,唤醒所有阻塞在屏障点的线程。在释放屏障点之后,可以先执行一个任务,再让所有阻塞被唤醒的线程继续之后后续任务。
Cyclic循环:所有线程被释放后,屏障点的数值可以再次被重置。
CyclicBarrier一般被称为栅栏。
CyclicBarrier是一种同步机制,允许一组线程互相等待。现成的达到屏障点其实是基于await方法在屏障点阻塞。
CyclicBarrier并没有基于AQS实现,他是基于ReentrantLock锁的机制去实现了对屏障点–,以及线程挂起的操作。(CountDownLatch本身是基于AQS,对state进行release操作后,可以-1)
CyclicBarrier没来一个线程执行await,都会对屏障数值进行-1操作,每次-1后,立即查看数值是否为0,如果为0,直接唤醒所有的互相等待线程。
CyclicBarrier对比CountDownLatch区别
- 底层实现不同。CyclicBarrier基于ReentrantLock做的。CountDownLatch直接基于AQS做的。
- 应用场景不同。CountDownLatch的计数器只能使用一次。而CyclicBarrier在计数器达到0之后,可以重置计数器。CyclicBarrier可以实现相比CountDownLatch更复杂的业务,执行业务时出现了错误,可以重置CyclicBarrier计数器,再次执行一次。
- CyclicBarrier还提供了很多其他的功能:
- 可以获取到阻塞的现成有多少
- 在线程互相等待时,如果有等待的线程中断,可以抛出异常,避免无限等待的问题。
- CountDownLatch一般是让主线程等待,让子线程对计数器–。CyclicBarrier更多的让子线程也一起计数和等待,等待的线程达到数值后,再统一唤醒
CyclicBarrier:多个线程互相等待,直到到达同一个同步点,再一次执行。
2、CyclicBarrier应用
出国旅游。
导游小姐姐需要等待所有乘客都到位后,发送护照,签证等等文件,再一起出发
比如Tom,Jack,Rose三个人组个团出门旅游
在构建CyclicBarrier可以指定barrierAction,可以选择性指定,如果指定了,那么会在barrier归0后,优先执行barrierAction任务,然后再去唤醒所有阻塞挂起的线程,并行去处理后续任务。
所有互相等待的线程,可以指定等待时间,并且在等待的过程中,如果有线程中断,所有互相的等待的线程都会被唤醒。
如果在等待期间,有线程中断了,唤醒所有线程后,CyclicBarrier无法继续使用。
如果线程中断后,需要继续使用当前的CyclicBarrier,需要调用reset方法,让CyclicBarrier重置。
如果CyclicBarrier的屏障数值到达0之后,他默认会重置屏障数值,CyclicBarrier在没有线程中断时,是可以重复使用的。
public static void main(String[] args) throws InterruptedException {
CyclicBarrier barrier = new CyclicBarrier(3,() -> {
System.out.println("等到各位大佬都到位之后,分发护照和签证等内容!");
});
new Thread(() -> {
System.out.println("Tom到位!!!");
try {
barrier.await();
} catch (Exception e) {
System.out.println("悲剧,人没到齐!");
return;
}
System.out.println("Tom出发!!!");
}).start();
Thread.sleep(100);
new Thread(() -> {
System.out.println("Jack到位!!!");
try {
barrier.await();
} catch (Exception e) {
System.out.println("悲剧,人没到齐!");
return;
}
System.out.println("Jack出发!!!");
}).start();
Thread.sleep(100);
new Thread(() -> {
System.out.println("Rose到位!!!");
try {
barrier.await();
} catch (Exception e) {
System.out.println("悲剧,人没到齐!");
return;
}
System.out.println("Rose出发!!!");
}).start();
/*
tom到位,jack到位,rose到位
导游发签证
tom出发,jack出发,rose出发
*/
}
3、CyclicBarrier源码分析
分成两块内容去查看,首先查看CyclicBarrier的一些核心属性,然后再查看CyclicBarrier的核心方法
(1)CyclicBarrier的核心属性
public class CyclicBarrier {
// 这个静态内部类是用来标记是否中断的
private static class Generation {
boolean broken = false;
}
/** CyclicBarrier是基于ReentrantLock实现的互斥操作,以及计数原子性操作 */
private final ReentrantLock lock = new ReentrantLock();
/** 基于当前的Condition实现线程的挂起和唤醒 */
private final Condition trip = lock.newCondition();
/** 记录有参构造传入的屏障数值,不会对这个数值做操作 */
private final int parties;
/** 当屏障数值达到0之后,优先执行当前任务 */
private final Runnable barrierCommand;
/** 初始化默认的Generation,用来标记线程中断情况 */
private Generation generation = new Generation();
/** 每来一个线程等待,就对count进行-- */
private int count;
}
(2)CyclicBarrier的有参构造
掌握构建CyclicBarrier之后,内部属性的情况
// 这个是CyclicBarrier的有参构造
// 在内部传入了parties,屏障点的数值
// 还传入了barrierAction,屏障点的数值达到0,优先执行barrierAction任务
public CyclicBarrier(int parties, Runnable barrierAction) {
// 健壮性判
if (parties <= 0) throw new IllegalArgumentException();
// 当前类中的属性parties是保存屏障点数值的
this.parties = parties;
// 将parties赋值给属性count,每来一个线程,继续count做-1操作。
this.count = parties;
// 优先执行的任务
this.barrierCommand = barrierAction;
}
(3)CyclicBarrier中的await方法
在CyclicBarrier中,提供了2个await方法
- 第一个是无参的方式,线程要死等,直屏障点数值为0,或者有线程中断
- 第二个是有参方式,传入等待的时间,要么时间到位了,要不就是直屏障点数值为0,或者有线程中断
无论是哪种await方法,核心都在于内部调用的dowait方法
dowait方法主要包含了线程互相等待的逻辑,以及屏障点数值到达0之后的操作
三、Semaphone应用&源码分析
1、Semaphore介绍
sync,ReentrantLock是互斥锁,保证一个资源同一时间只允许被一个线程访问
Semaphore(信号量)保证1个或多个资源可以被指定数量的线程同时访问
底层实现是基于AQS去做的。
Semaphore底层也是基于AQS的state属性做一个计数器的维护。state的值就代表当前共享资源的个数。如果一个线程需要获取的1或多个资源,直接查看state的标识的资源个数是否足够,如果足够的,直接对state - 1拿到当前资源。如果资源不够,当前线程就需要挂起等待。知道持有资源的线程释放资源后,会归还给Semaphore中的state属性,挂起的线程就可以被唤醒。
Semaphore也分为公平和非公平的概念。
使用场景:连接池对象就可以基础信号量去实现管理。在一些流量控制上,也可以采用信号量去实现。再比如去迪士尼或者是环球影城,每天接受的人流量是固定的,指定一个具体的人流量,可能接受10000人,每有一个人购票后,就对信号量进行–操作,如果信号量已经达到了0,或者是资源不足,此时就不能买票。
2、Semaphore应用
以上面环球影城每日人流量为例子去测试一下。
public static void main(String[] args) throws InterruptedException {
// 今天环球影城还有人个人流量
Semaphore semaphore = new Semaphore(10);
new Thread(() -> {
System.out.println("一家三口要去~~");
try {
semaphore.acquire(3);
System.out.println("一家三口进去了~~~");
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println("一家三口走了~~~");
semaphore.release(3);
}
}).start();
for (int i = 0; i < 7; i++) {
int j = i;
new Thread(() -> {
System.out.println(j + "大哥来了。");
try {
semaphore.acquire();
System.out.println(j + "大哥进去了~~~");
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
System.out.println(j + "大哥走了~~~");
semaphore.release();
}
}).start();
}
Thread.sleep(10);
System.out.println("main大哥来了。");
if (semaphore.tryAcquire()) {
System.out.println("main大哥进来了。");
}else{
System.out.println("资源不够,main大哥进来了。");
}
Thread.sleep(10000);
System.out.println("main大哥又来了。");
if (semaphore.tryAcquire()) {
System.out.println("main大哥进来了。");
semaphore.release();
}else{
System.out.println("资源不够,main大哥进来了。");
}
}
其实Semaphore整体就是对构建Semaphore时,指定的资源数的获取和释放操作
获取资源方式:
- acquire():获取一个资源,没有资源就挂起等待,如果中断,直接抛异常
- acquire(int):获取指定个数资源,资源不够,或者没有资源就挂起等待,如果中断,直接抛异常
- tryAcquire():获取一个资源,没有资源返回false,有资源返回true
- tryAcquire(int):获取指定个数资源,没有资源返回false,有资源返回true
- tryAcquire(time,unit):获取一个资源,如果没有资源,等待time.unit,如果还没有,就返回false
- tryAcquire(int,time,unit):获取指定个数资源,如果没有资源,等待time.unit,如果还没有,就返回false
- acquireUninterruptibly():获取一个资源,没有资源就挂起等待,中断线程不结束,继续等
- acquireUninterruptibly(int):获取指定个数资源,没有资源就挂起等待,中断线程不结束,继续等
归还资源方式:
- release():归还一个资源
- release(int):归还指定个数资源
3、Semaphore源码分析
先查看Semaphore的整体结构,然后基于获取资源,以及归还资源的方式去查看源码
(1)Semaphore的整体结构
Semaphore内部有3个静态内类。
首先是向上抽取的Sync
其次还有两个Sync的子类NonFairSync以及FairSync两个静态内部类
Sync内部主要提供了一些公共的方法,并且将有参构造传入的资源个数,直接基于AQS提供的setState方法设置了state属性。
NonFairSync以及FairSync区别就是tryAcquireShared方法的实现是不一样。
(2)Semaphore的非公平的获取资源
在构建Semaphore的时候,如果只设置资源个数,默认情况下是非公平。
如果在构建Semaphore,传入了资源个数以及一个boolean时,可以选择非公平还是公平。
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
从非公平的acquire方法入手
首先确认默认获取资源数是1个,并且acquire是允许中断线程时,抛出异常的。获取资源的方式,就是直接用state - 需要的资源数,只要资源足够,就CAS的将state做修改。如果没有拿到锁资源,就基于共享锁的方式去将当前线程挂起在AQS双向链表中。如果基于doAcquireSharedInterruptibly拿锁成功,会做一个事情。会执行setHeadAndPropagate方法。一会说
// 信号量的获取资源方法(默认获取一个资源)
public void acquire() throws InterruptedException {
// 跳转到了AQS中提供共享锁的方法
sync.acquireSharedInterruptibly(1);
}
// AQS提供的
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
// 判断线程的中断标记位,如果已经中断,直接抛出异常
if (Thread.interrupted())
throw new InterruptedException();
// 先看非公平的tryAcquireShared实现。
// tryAcquireShared:
// 返回小于0,代表获取资源失败,需要排队。
// 返回大于等于0,代表获取资源成功,直接执行业务代码
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 信号量的非公平获取资源方法
final int nonfairTryAcquireShared(int acquires) {
// 死循环。
for (;;) {
// 获取state的数值,剩余的资源个数
int available = getState();
// 剩余的资源个数 - 需要的资源个数
int remaining = available - acquires;
// 如果-完后,资源个数小于0,直接返回这个负数
if (remaining < 0 ||
// 说明资源足够,基于CAS的方式,将state从原值,改为remaining
compareAndSetState(available, remaining))
return remaining;
}
}
// 获取资源失败,资源不够,当前线程需要挂起等待
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
// 构建Node节点,线程和共享锁标记,并且到AQS双向链表中
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
// 拿到上一个节点
final Node p = node.predecessor();
// 如果是head.next,就抢一手
if (p == head) {
// 再次基于非公平的方式去获取一次资源
int r = tryAcquireShared(arg);
// 到这,说明拿到了锁资源
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null;
failed = false;
return;
}
}
// 如果上面没拿到,或者不是head的next节点,将前继节点的状态改为-1,并挂起当前线程
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
// 如果线程中断会抛出异常
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
acquire()以及acquire(int)的方式,都是执行acquireSharedInterruptibly方法去尝试获取资源,区别只在于是否传入了需要获取的资源个数。
tryAcquire()以及tryAcquire(int因为这两种方法是直接执行tryAcquire,只使用非公平的实现,只有非公平的情况下,才有可能在有线程排队的时候获取到资源
但是tryAcquire(int,time,unit)这种方法是正常走的AQS提供的acquire。因为这个tryAcquire可以排队一会,即便是公平锁也有可能拿到资源。这里的挂起和acquire挂起的区别仅仅是挂起的时间问题。
- acquire是一直挂起直到线程中断,或者线程被唤醒。
- tryAcquire(int,time,unit)是挂起一段时间,直到线程中断,要么线程被唤醒,要么阻塞时间到了
还有acquireUninterruptibly()以及acquireUninterruptibly(int)只是在挂起线程后,不会因为线程的中断而去抛出异常
(3)Semaphore公平实现
公平与非公平只是差了一个方法的实现tryAcquireShared实现
这个方法的实现中,如果是公平实现,需要先查看AQS中排队的情况
// 信号量公平实现
protected int tryAcquireShared(int acquires) {
// 死循环。
for (;;) {
// 公平实现在走下述逻辑前,先判断队列中排队的情况
// 如果没有排队的节点,直接不走if逻辑
// 如果有排队的节点,发现当前节点处在head.next位置,直接不走if逻辑
if (hasQueuedPredecessors())
return -1;
// 下面这套逻辑和公平实现是一模一样的。
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
(4)Semaphore释放资源
因为信号量从头到尾都是共享锁的实现……
释放资源操作,不区分公平和非公平
// 信号量释放资源的方法入口
public void release() {
sync.releaseShared(1);
}
// 释放资源不分公平和非公平,都走AQS的releaseShared
public final boolean releaseShared(int arg) {
// 优先查看tryReleaseShared,这个方法是信号量自行实现的。
if (tryReleaseShared(arg)) {
// 只要释放资源成功,执行doReleaseShared,唤醒AQS中排队的线程,去竞争Semaphore的资源
doReleaseShared();
return true;
}
return false;
}
// 信号量实现的释放资源方法
protected final boolean tryReleaseShared(int releases) {
// 死循环
for (;;) {
// 拿到当前的state
int current = getState();
// 将state + 归还的资源个数,新的state要被设置为next
int next = current + releases;
// 如果归还后的资源个数,小于之前的资源数。
// 避免出现归还资源后,导致next为负数,需要做健壮性判断
if (next < current)
throw new Error("Maximum permit count exceeded");
// CAS操作,保证原子性,只会有一个线程成功的就之前的state修改为next
if (compareAndSetState(current, next))
return true;
}
}
4、AQS 中PROPAGATE 节点
为了更好的了解PROPAGATE节点状态的意义,优先从JDK1.5去分析一下释放资源以及排队后获取资源的后置操作
(1)掌握JDK1.5-Semaphore执行流程图
首先查看4个线程获取信号量资源的情况
往下查看释放资源的过程会触发什么问题
首先t1释放资源,做了进一步处理
当线程3获取锁资源后,线程2再次释放资源,因为执行点问题,导致线程4无法被唤醒
(2)分析JDK1.8的变化
====================================JDK1.5实现============================================.
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
private void setHeadAndPropagate(Node node, int propagate) {
setHead(node);
if (propagate > 0 && node.waitStatus != 0) {
Node s = node.next;
if (s == null || s.isShared())
unparkSuccessor(node);
}
}
====================================JDK1.8实现============================================.
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
private void doReleaseShared() {
for (;;) {
// 拿到head节点
Node h = head;
// 判断AQS中有排队的Node节点
if (h != null && h != tail) {
// 拿到head节点的状态
int ws = h.waitStatus;
// 状态为-1
if (ws == Node.SIGNAL) {
// 将head节点的状态从-1,改为0
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 唤醒后继节点
unparkSuccessor(h);
}
// 发现head状态为0,将head状态从0改为-3,目的是为了往后面传播
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 没有并发的时候。head节点没变化,正常完成释放排队的线程
if (h == head)
break;
}
}
private void setHeadAndPropagate(Node node, int propagate) {
// 拿到head
Node h = head;
// 将线程3的Node设置为新的head
setHead(node);
// 如果propagate 大于0,代表还有剩余资源,直接唤醒后续节点,如果不满足,也需要继续往后判断看下是否需要传播
// h == null:看成健壮性判断即可
// 之前的head节点状态为负数,说明并发情况下,可能还有资源,需要继续向后唤醒Node
// 如果当前新head节点的状态为负数,继续释放后续节点
if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
// 唤醒当前节点的后继节点
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}