并发编程基础
目录
- 并发编程
- Java实现线程的三种方式
- 线程的生命周期
- API
- Thread.join()
- Thread.sleep()
- wait()和notify()
- Thread.interrupted和Thread.interrupt
- 线程安全性
- 并发编程问题的源头-原子性、可见性、有序性
- JMM
- Synchronized
- synchronized锁的范围
- synchronized的本质
- volatile
- final
- Happens-Before
- 原子类Atomic-无锁工具的典范
- Atomic实现原理
- CAS：compareAndSwapInt(object, offset, expect, update)
- Atomic分类
- ThreadLocal的使用和原理
- 发布与逃逸
- J.U.C之AQS
- AQS(AbstractQueuedSynchronizer)是什么？
- ReentrantLock重入锁
- CountDownLatch计时器
- Semaphore信号灯
- CyclicBarrier栅栏
- Condition
并发编程
线程调度算法：
操作系统中，CPU竞争有很多种策略。Unix系统使用的是时间片算法，而Windows则属于抢占式的。
时间片算法：
所有的进程都排成一个队列，操作系统按照他们的顺序给每个进程分配一段时间，如果在时间片结束时还在运行，则会把这个进程阻塞，然后把CPU时间片分配给其他进程。
抢占式：
如果一个进程得到了CPU时间，除非自己放弃使用，否则将完整的霸占CPU。
Java实现线程的三种方式
ThreadDemo、RunnableDemo、CallableDemo
public class ThreadDemo extends Thread {
@Override
public void run() {
System.out.println("当前线程：" + Thread.currentThread().getName());
}
public static void main(String[] args) {
ThreadDemo threadDemo = new ThreadDemo();
threadDemo.start();
}
}
public class RunnableDemo implements Runnable {
@Override
public void run() {
System.out.println("当前线程：" + Thread.currentThread().getName());
}
public static void main(String[] args) {
new Thread(new RunnableDemo()).start();
}
}
public class CallableDemo implements Callable {
@Override
public String call() throws Exception {
return "当前线程：" + Thread.currentThread().getName();
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(1);
Future future = executorService.submit(new CallableDemo());
System.out.println(future.get() // future.get()是一个阻塞方法，必须等到Callable.call()给到返回值
);
}
}
线程的生命周期
Java线程从创建到销毁，一共经历6个状态
- new：初始状态，线程被构建，但是还没有调用start()方法
- runnabled：运行状态，Java线程把操作系统中的就绪和运行两种状态统一称为“运行中”，就绪过程是等待CPU调度。
- blocked：阻塞状态，表示线程进入等待状态，也就是线程因为某种原因放弃了CPU使用权，阻塞也分为几种情况
- waiting：等待状态
- time_waiting：超时等待状态，超时以后自动返回
- terminated：终止状态，表示当前线程执行完毕。
// 当runnable状态时，调用了
Thread.sleep(long);
Object.wait(long);
Thread.join(long);
LockSupport.parkNanos(long);
LockSupport.parkUntil(long);
// 会进入超时等待状态，当long时间到达，会自动唤醒，也可通过以下方法主动唤醒
Object.notify();
Object.notifyAll();
LockSuppoort.unpark();
// 当runnable状态时，调用了
Object.wait();
Object.join();
LockSupport.park();
// 会进入等待状态，通过以下方法主动唤醒
Object.notify();
Object.notifyAll();
LockSupport.unpark(Thread);
// 进入blocked状态只有一种情况，等待进入synchronized方法或代码块
// 当获取到同步锁会进入runnable状态
API
Thread.join()
Thread.join的作用是保证线程执行结果的可见性。阻塞调用者所在的线程。
join()
public final synchronized void join(long millis) throws InterruptedException {
long base = System.currentTimeMillis();
long now = 0;
if (millis < 0) {
throw new IllegalArgumentException("timeout value is negative");
}
if (millis == 0) {
while (isAlive()) { // 线程循环调用isAlive()会一直阻塞主线程，如果非isAlive()会被jvm唤醒
wait(0);
}
} else {
while (isAlive()) {
long delay = millis - now;
if (delay <= 0) {
break;
}
wait(delay);
now = System.currentTimeMillis() - base;
}
}
}
Thread.sleep()
使线程暂停执行一段时间，直到等待的时间结束才恢复执行或在这段时间内被中断。
Thread.sleep的工作流程：
挂起线程并修改其运行状态
用sleep()提供的参数来设置一个定时器。
当时间结束，定时器会触发，内核收到中断后修改线程的运行状态。
例如：线程会被标志为就绪而进入就绪队列等待调度。
问题思考：
-
假设现在是2022-03-08 12:00:00.000，如果我调用一下Thread.sleep(1000)，在2022-03-08 12:00:01.000的时候，这个线程会不会唤醒？
不一定。如果1000ms过去后，CPU正在处理其他线程，操作系统是不会重新分配CPU给这个线程的，直到那个线程挂起或者结束；如果恰巧这个时候轮到操作系统分配CPU，那么当前线程也不一定是优先级最高的，CPU还可能被其他线程抢占过去。
-
Thread.sleep(0)的意义。
类似Thread.yield()，出让CPU，触发操作系统重新进行CPU竞争。Thread.sleep(0)就是告诉操作系统在未来的0ms内，我不想参与CPU竞争，这个操作指令会被操作系统接受，于是这个时候操作系统会重新计算大家总的优先级，然后重新分配CPU资源。
wait()和notify()
思考：如何实现一个线程修改了一个对象的值，而另一个线程感知到了变化，然后进行响应的操作。
点击查看代码
// 生产者 public class Procedure implements Runnable { private final Queuebags; private int size; public Procedure(Queue bags, int size) { this.bags = bags; this.size = size; } @Override public void run() { int i = 0; while (true) { i++; synchronized (bags) { while (bags.size() == size) { System.out.println("bags已经满了"); try { bags.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("生产者生产：bag" + i); bags.add("bag" + i); bags.notifyAll(); } } } } // 消费者 public class Consumer implements Runnable { private final Queue bags; private int size; public Consumer(Queue bags, int size) { this.bags = bags; this.size = size; } @Override public void run() { while (true) { synchronized (bags) { while (bags.isEmpty()) { System.out.println("bags为空"); try { bags.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } String bag = bags.remove(); System.out.println("消费者消费：" + bag); bags.notifyAll(); } } } } // Test public class WaitNotifyDemo { public static void main(String[] args) { Queue queue = new LinkedList<>(); int size = 10; Procedure procedure = new Procedure(queue, size); Consumer consumer = new Consumer(queue, size); Thread t1 = new Thread(procedure); Thread t2 = new Thread(consumer); t1.start(); t2.start(); } }
为什么wait/notify需要加synchronized？
- 其实wait/notify本质上其实是一种条件的竞争，wait和notify方法一定是互斥存在的，既然要实现互斥，那么synchronized就是一个很好的解决办法
- wait和notify用于实现多个线程之间的通信，而通信必然存在一个通信的载体。wait/notify是基于synchronized来实现通信的。也就是两者必须要在同一个频道(也就是同一个锁)内
Thread.interrupted和Thread.interrupt
如何正确终止一个线程？
- 通过共享变量的值让一段代码终止
- interrupt方法
interrupt方法
当其他线程通过调用当前线程的interrupt方法，表示向当前线程打个招呼，告诉他可以中断线程的执行了，至于什么时候中断，取决于当前线程自己
public class InterruptDemo {
private static int i;
public static void main(String[] args) throws InterruptedException {
Thread thread = new Thread(() -> {
// Thread.currentThread().isInterrupted()默认是false
while (!Thread.currentThread().isInterrupted()) {
i++;
}
});
thread.start();
TimeUnit.SECONDS.sleep(2);
// interrupt这个属性 false -> true
thread.interrupt();
System.out.println(i);
}
}
什么时候调用中断？
- while()
- 线程处于阻塞状态的情况下，中断才有意义
- join()
- wait()
- Thread.sleep()
interrupted()
Thread.interrupted()对设置中断标识的线程复位，并返回当前的中断状态。让线程恢复成未被中断状态。
线程安全性
并发编程问题的源头-原子性、可见性、有序性
如何理解线程安全？
当多个线程访问某个对象时，不管运行时环境采用何种调度方式或者这些线程将如何交替执行，并且在主调代码中不需要任何额外的同步或者协同，这个类都能表现出正确的行为，那么就称这个类是线程安全的。
线程安全问题的本质：
- 原子性
- 可见性
- 有序性
CPU的高速缓存和多核CPU，操作系统增加了进程，线程以及分时复用CPU，均衡CPU与I/O设备的速度差异，编译程序优化指令的执行顺序，使得能够更加合理的利用缓存。
-
可见性：
public class VisableDemo { public static boolean stop = false; public static void main(String[] args) throws InterruptedException { Thread thread = new Thread(() -> { int i = 0; while (!stop) { i ++; } System.out.println("result:" + i); }); thread.start(); System.out.println("begin start thread"); Thread.sleep(1000); stop=true; } } // 运行结束后发现并没有对子线程产生任何影响，说明主线程对stop的操作，子线程不可见。 -
原子性：
线程A将count=0加载到寄存器，切换线程，线程B也将count = 0加载到寄存器，两个线程同时执行了count++，分别写入到了内存。
-
有序性(指令重排)
线程A执行a = 1, x = b，x初始值为0，线程B执行b = 2, y = a，y初始值为0；那么很有可能两个线程执行完，x和y都为0。
JMM
Java内存模型是一种抽象结构，它提供了合理的禁用缓存以及禁止重排序的方法来解决可见性、有序性问题。
可见性、有序性的解决方案
- volatile、synchronized、final关键字
- Happens-Before原则
JMM并没有限制执行引擎使用处理器的寄存器或者高速缓存来提升指令的执行速度，也没有去限制编译器对指令进行重排序，也就是说在JMM中，他会存在缓存一致性问题和指令重排序问题，只是JMM把底层的问题抽象到了JVM层面，再基于CPU层面提供的内存屏障指令，以及限制编译器重排序的一些操作来解决并发问题。
Synchronized
synchronized锁的范围
- 对于普通同步方法，锁是当前实例对象
- 对于静态同步方法，锁是当前类的Class对象。
- 对于同步方法快，锁是synchronized括号里配置的对象。
public class SynchronizedDemo {
// 对象锁(同一个对象内有效)
public synchronized void demo() {
}
public void demo1() {
synchronized (this) { // 对象锁
// 范围可控，只有这里需要加锁
}
}
public void demo2() {
synchronized (SynchronizedDemo.class) { // 类级别的锁
// 范围可控，只有这里需要加锁
}
}
// 类级别锁
public synchronized static void demo3() {
}
public static void main(String[] args) {
SynchronizedDemo syncDemo1 = new SynchronizedDemo();
SynchronizedDemo syncDemo2 = new SynchronizedDemo();
// 如果都访问dmeo3的话客户以实现线程的互斥
new Thread(() -> {
syncDemo1.demo();
}).start();
new Thread(() -> {
syncDemo1.demo();// 可以实现两个线程的互斥
syncDemo2.demo();// 无法实现两个线程的互斥
}).start();
}
}
要做到互斥，就必须要争夺共享资源
synchronized的本质
monitorenter：获得锁
monitorexit：释放锁(正常情况/异常情况)
monitor：对对象监视器的获取
两个线程同时访问加了锁的代码，只有一个线程能获取到这个临界资源，没有获得monitor的线程会加入到同步队列，就处于阻塞状态，直到调用了monitorexit释放锁后，同步队列里的线程会被唤醒，会再去争抢这个监视器。
这个监视器它是依赖操作系统里面的mutex lock互斥锁来实现的，线程阻塞后便进到了内核的调度状态，这会导致系统在用户态，内核态一个来回切换，会影响性能。在1.6之前是一个重量级锁。为了保证安全性和性能，在jdk1.6之后做了一些优化：
- 自适应自旋锁
- 引入偏向锁，轻量级锁
- 锁消除、锁粗化
volatile
volatile可以用来解决可见性和有序性问题。
public class VisableDemo {
public volatile static boolean stop = false;
public static void main(String[] args) throws InterruptedException {
Thread thread = new Thread(() -> {
int i = 0;
while (!stop) {
i ++;
}
System.out.println("result:" + i);
});
thread.start();
System.out.println("begin start thread");
Thread.sleep(1000);
stop=true;
}
}
当加了volatile之后，会在编译过后执行底层的Lock指令
Lock指令的作用：
- 将当前处理器缓存行的数据写回到系统内存
- 这个写回内存的操作会使在其他CPU里缓存了该内存地址的数据无效
什么情况下需要用到volatile
当存在多个线程对同一个共享变量进行修改的时候，需要增加volatile，保证数据修改的实时可见。
volatile如何解决有序性问题的？
要解决这个问题，首先了解CPU层面的内存屏障
- Sotre Barrier：强制所有在store屏障指令之前的store指令，都在该store屏障指令执行之前被执行，并把store缓冲区的数据都刷到CPU缓存
- Load Barrier：强制所有在load屏障指令之后的load指令，都在该load屏障指令执行之后被执行，并且一直等到load缓冲区被该CPU读完才能执行之后的load指令
- Full Barrier：复合了load和store屏障的功能
Java提供了4种屏障：
volatile的总结
本质上来说：volatile实际上是通过内存屏障来防止指令重排序以及禁止CPU高速缓存来解决可见性问题。
而#Lock指令，它本意上是禁止高速缓存解决可见性问题，但实际上在这里，它表示的是一种内存屏障的功能。也就是说针对当前的硬件环境，JMM层面采用Lock指令作为内存屏障来解决可见性问题。
final
final关键字
final在Java中是一个保留的关键字，可以声明成员变量、方法、类以及本地变量。一旦你将引用声明作final，你将不能改变这个引用了。
final域和线程安全有什么关系？
对于final域，编译器和处理器要遵守两个重排序规则。可以基于重排序的规则来解决可见性问题。
在构造函数内对一个final域的写入，与随后把这个被构造对象的引用赋值给另一个引用变量，这两个操作之间不能重排序。
初次读一个包含final域的对象的引用，与随后初次读这个final域，这两个操作之间不能重排序。
写final域重排序规则
JMM禁止编译器把final域的写重排序到构造函数之外
编译器会在final域的写之后，构造函数return之前，插入一个StoreStore屏障。这个屏障禁止处理器把final域的写重排序到构造函数之外。
读final域重排序规则
- 在一个线程中，初次读对象引用与初次读该对象包含的final域，JMM禁止处理器重排序这两个操作，编译器会在读final域操作的前面插入一个LoadLoad屏障。
public class FinalDemo {
int i;
final int j;
static FinalDemo obj;
public FinalDemo() {
i = 1; // 可能在这个构造方法之外执行
j = 2; // 只能在构造函数中执行
}
public static void writer() {
obj = new FinalDemo();
}
public static void reader() {
FinalDemo object = obj; // 可能会在初始化之前执行
int a = object.i; // 可能读到object.i = 0或者nullpointerException，可能在这里还没引用
int b = object.j;
}
public static void main(String[] args) {
new Thread(FinalDemo::writer).start();
new Thread(FinalDemo::reader).start();
}
}
溢出带来的重排序问题
public class FinalReferenceEscapeDemo {
final int i;
static FinalReferenceEscapeDemo obj;
public FinalReferenceEscapeDemo() {
i = 1; // 写final域
obj = this; // this引用“逃逸”，要满足final的规则，就不允许逃逸的产生
}
public static void writer() {
new FinalReferenceEscapeDemo();
}
public static void reader() {
if(obj != null) {
int temp = obj.i;
}
}
}
Happens-Before
什么是Happens-Before
Happens-Before是一种可见性规则，它表达的含义是前面一个操作的结果对后续操作是可见的。
6种Happens-Before规则，不用再去考虑可见性问题
- 程序顺序规则：单线程程序执行结果是不能改变的as-if-serial
- 监视器所规则：对一个锁的解锁一定在后续对这个锁加锁之前
- volatile变量规则：对一个volatile域的写，一定在任意后续对这个volatile域的读之前(实际上是基于volatile域的store-load Barrie)
- 传递性：if A Happens-Before B, and B Happens-Before C, then A Happens-Before C
- start()规则：如果A线程执行操作ThreadB.start()(启动B线程)，那么A线程的ThreadB.start()操作一定在B线程的任意操作之前。
- join()规则：如果线程A执行操作ThreadB.join()并成功返回，那么线程B中的任意操作一定在A线程从ThreadB.join()成功返回之前。
原子类Atomic-无锁工具的典范
原子性实现方式：
- synchronized、Lock
- J.U.C包下的Atomic类
synchronized写法
public class AtomicDemo {
public static int count = 0;
public synchronized static void incr() {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
count++;
}
public static void main(String[] args) throws InterruptedException {
for ( int i = 0; i < 1000; i++) {
new Thread(AtomicDemo::incr).start();
}
Thread.sleep(4000);
System.out.println("result:" + count);
}
}
Atomic写法
public class AtomicDemo {
// public static int count = 0;
private static AtomicInteger atomicInteger = new AtomicInteger(0);
public static void incr() {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
//count++;
atomicInteger.incrementAndGet();
}
public static void main(String[] args) throws InterruptedException {
for ( int i = 0; i < 1000; i++) {
new Thread(AtomicDemo::incr).start();
}
Thread.sleep(4000);
// System.out.println("result:" + count);
System.out.println("result：" + atomicInteger.get());
}
}
Atomic实现原理
- Unsafe类
- CAS
CAS：compareAndSwapInt(object, offset, expect, update)
- object：当前atomic对象
- offset：内存中的偏移量
- expect：预期的值
- update：如果内存中的偏移量对应的值等于预期值，那么就修改成update
public final int getAndAddInt(Object var1, long var2, int var4) {
int var5;
do {
var5 = this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
return var5;
}
实际上是基于乐观锁，只有相同的时候才改变值，对比Synchronized是一个内核态与用户态一直在转化，让线程处于阻塞，CAS一直运行循环的性能损耗是很低的。
Atomic分类
- 原子更新基本类型
- 原子更新数组
- 原子更新引用类型
- 原子更新字段类
ThreadLocal的使用和原理
ThreadLocal实际上是一种线程安全的解决方案之一，有点类似于一个线程的副本，就是说当多个线程并行去访问某个共享变量进行修改的时候，可以对每个线程去进行隔离，从而去解决并发数据安全的问题。
public class ThreadLocalDemo {
private static Integer num = 0;
public static final ThreadLocal local = new ThreadLocal(){
protected Integer initialValue() {
return 0; // 初始值
}
};
public static void main(String[] args) {
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
// num += 5;
num = local.get(); // 拿到初始值
num += 5;
local.set(num);
System.out.println(Thread.currentThread().getName() + "->" + num);
}, "Thread-" + i);
}
for (Thread thread : threads) {
thread.start();
}
}
}
ThreadLocal
public class ThreadLocal {
public T get() {
Thread t = Thread.currentThread();
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();
}
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
map.set(this, value);
} else {
createMap(t, value);
}
}
static class ThreadLocalMap {
static class Entry extends WeakReference> {
/** The value associated with this ThreadLocal. */
Object value;
Entry(ThreadLocal<?> k, Object v) {
super(k);
value = v;
}
}
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
table = new Entry[INITIAL_CAPACITY];
int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
table[i] = new Entry(firstKey, firstValue);
size = 1;
setThreshold(INITIAL_CAPACITY);
}
}
}
在Thread中，有一个ThreadLoacl.ThreadLocalMap属性，将Thread与ThreadLocalMap绑定。
在ThreadLocal.get()/set()方法中，将Thread中的ThreadLocalMap属性设置Key为ThreadLocal，将value设置为value。
保证了每个Thread中的ThreadLocalMap虽然是不同对象，但是ThreadLocal是相同的。
可以创建多个不同初始值的ThreadLocal对象，ThreadLocalMap中维护了一个Entry数组，这个数组可以存储不同ThreadLocal对应的键值。
发布与逃逸
发布的意思是使一个对象能够被当前范围之外的代码所使用。
不安全发布：
private String[] states = {"a", "b", "c", "d"};
// 发布出去一个
public String[] getStates() {
return states;
}
public static void main(String[] args){
App unSafePub = new App();
System.out.println("Init array is " + Arrays.toString(unSafePub.getStates()));
unSafePub.getStates()[0] = "Elian";
System.out.println("After modify.the array is " + Arrays.toString(unSafePub.getStates()));
}
对象溢出(逃逸)是指一种错误的发布，当一个对象还没有构造完成时，就使它被其他线程所见。
安全发布对象的四种方法：
- 在静态初始化函数中初始化一个对象引用
- 将对象的引用保存到volatile类型的域或者AtomicReference对象中(利用volatile happen-before原则)
- 将对象的引用保存到某个正确构造对象的final类型域中(初始化安全性)
- 将对象的引用保存到一个有锁保护的域中(读写都上锁)
public class StaticDemo {
private StaticDemo() {
}
private static StaticDemo instance = new StaticDemo();
public static StaticDemo getInstance() {
return instance;
}
}
public class FinalAreaDemo {
private final Map status;
public FinalAreaDemo() {
status = new HashMap<>();
status.put("name", "elian");
}
}
public class VolatileSyncDemo {
private VolatileSyncDemo() {
}
private static volatile VolatileSyncDemo instance = null;
// DCL问题：double check lock
/**
* instance = new VolatileSyncDemo
* 在jvm层面有三个步骤
* 1. memory = allocate() 分配内存
* 2. ctorInstance(memory) 创建对象
* 3. instance = memory 将对象指向分配的内存
* 2和3是可以进行重排序的
* 如果1.3.2的顺序初始化实例，可能导致instance不为null，但实际上是没有初始化完成的，
* 所以在instance前加上volatile，就可以避免指令重排序带来的不完整对象了
*/
public static VolatileSyncDemo getInstance() {
if (instance == null) {
synchronized (VolatileSyncDemo.class) { // 缩小锁的范围，只有instance为空的时候才锁
if (instance == null)
instance = new VolatileSyncDemo();
}
}
return instance;
}
}
J.U.C之AQS
AQS(AbstractQueuedSynchronizer)是什么？
AQS本身是一个同步工具，也是Lock实现同步的一个核心组件，用来实现线程的排队阻塞操作。
ReentrantLock重入锁
什么是重入锁？
一个持有锁的线程，在释放锁之前，如果再次访问加了该同步锁的其他方法，这个线程不需要再次争抢锁，只需要记录重入次数。
public class LockDemo {
static Lock lock = new ReentrantLock(); // 重入锁
public static int count = 0;
public static void incr() {
try {
lock.lock(); // 获得锁
Thread.sleep(1);
decr(); // 不需要争抢锁，只需要记录获得锁的次数，避免了死锁
count++;
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock(); // 释放锁
}
}
public static void decr() {
lock.lock();
count--;
lock.unlock();
}
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 1000; i++) {
new Thread(LockDemo::incr).start();
}
Thread.sleep(4000);
System.out.println("result:" + count);
}
}
ReentrantLock
public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer {
abstract void lock();
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread(); // 获取当前线程
int c = getState(); // 获取AQS的state
if (c == 0) { // 如果state是没被占用，继续重复非公平锁占锁操作
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;
}
protected final boolean tryRelease(int releases) { // AQS tryRelease
int c = getState() - releases; // lock 状态置为 0
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null); // 当前占用lock的锁置为空
}
setState(c); // 设置lock状态
return free;
}
}
public ReentrantLock() {
sync = new NonfairSync();
}
static final class NonfairSync extends Sync { // 非公平锁
private static final long serialVersionUID = 7316153563782823691L;
final void lock() {
if (compareAndSetState(0, 1)) // CAS原子操作，乐观锁，预期值0，修改值1，只有一个线程能获取到锁CAS成功的状态
setExclusiveOwnerThread(Thread.currentThread()); // 设置当前线程为独占线程AOS.thread = currThread
else
acquire(1); // AQS acquire(arg)
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
public void lock() {
sync.lock();
}
public void unlock() {
sync.release(1); // AQS
}
}
//AQS
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
private volatile int state;
static final class Node {
}
private Node addWaiter(Node mode) { // 双向链表，维护所有抢占锁的线程
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
public final void acquire(int arg) {
if (!tryAcquire(arg) && // Sync中的nonfairTryAcquire() 抢占锁
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 维护线程，并设置为可唤醒
selfInterrupt();
}
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) { // 自旋
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) { // 再次抢占锁
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) && // 把Node状态变为可唤醒，返回false，全部线程改变完再返回true
parkAndCheckInterrupt()) // 完成阻塞
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else { // == 0
compareAndSetWaitStatus(pred, ws, Node.SIGNAL); // 把Node状态变为-1，可唤醒状态
}
return false;
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 挂起阻塞当前的线程 通过onpark唤醒
return Thread.interrupted();
}
public final boolean release(int arg) { // ReenTrantLock.unlock //
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0); // 设置争夺状态
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread); // 唤醒head的next线程，only one，非公平锁
}
}
ReenTrantLock实现原理，在AQS中维护一个state，再进入的线程看到state为0，那么会加入阻塞队列，否则会尝试先去抢占资源(非公平锁)，如果lock释放锁unlock，会在阻塞链表中拿出一个线程争夺资源，并将上一个资源交给设为null，等待GC处理。同时在AOS里也会维护一个Thread属性，该属性保证了如果是同一个线程进入同一把锁，不会让该线程去争夺锁，而是直接记录争夺锁的次数，避免了死锁。
CountDownLatch计时器
countDownLatch是一个同步工具类，它允许一个或多个线程一直等待，知道其他线程的操作执行完毕再执行。从命名可以解读到countDown是倒数的意思，类似于倒计时的概念。
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(3);
new Thread(()->{
countDownLatch.countDown(); // 倒计时 3-1 = 2
}).start();
new Thread(()->{
countDownLatch.countDown(); // 倒计时 2-1 = 1
}).start();
new Thread(()->{
countDownLatch.countDown(); // 倒计时 1-1 = 0 -> 触发唤醒操作
}).start();
countDownLatch.await(); // 阻塞主线程
System.out.println("线程执行完毕");
}
类似Thread.join()方法。
用countDownLatch测试并发：
public class CountDownLatchDemo implements Runnable {
static CountDownLatch countDownLatch = new CountDownLatch(1);
public static void main(String[] args) {
for (int i = 0; i < 1000; i++) {
new Thread(new CountDownLatchDemo()).start(); // 构建1000个线程都在start状态
}
countDownLatch.countDown();
}
@Override
public void run() {
try {
countDownLatch.await(); // 所有线程都在等待countDownLatch.countDown后执行sth.
// TODO
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
源码分析
public class CountDownLatch {
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
setState(count); // AQS state存储3
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
private final Sync sync;
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1); //
}
public void countDown() {
sync.releaseShared(1); // 调用AQS
}
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0) // getState() > 0 ? 1 : -1
doAcquireSharedInterruptibly(arg); //
}
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
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); // 把当前节点设置为head节点，完成唤醒传递
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) && // 把Node状态变为可唤醒，返回false，全部线程改变完再返回true
parkAndCheckInterrupt()) // 完成阻塞
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) { // sync：state - 1
doReleaseShared();
return true;
}
return false;
}
private void doReleaseShared() { // 唤醒线程
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
}
CountDownLatch会在AQS维护一个state作为共有几个线程，当一个线程执行了await时，会将该线程加入AQS阻塞队列，直到countDown执行减1操作，state为0时，唤醒阻塞中的线程，并且会自旋的唤醒链表中阻塞的下一个线程。
Semaphore信号灯
semaphore也就是我们常说的信号灯，semaphore可以控制同时访问的线程个数，通过acquire获取一个许可，如果没有就等待，通过release释放一个许可。有点类似先流的作用。叫信号灯的原因也和它的用处有关，比如某商场就5个停车位，每个停车位只能听一辆车，如果这个时候来了10辆车，必须要等前面有空的车位才能进入。
原理：Semaphore在AQS维护了一个state，能够执行最大线程数，当有其他线程抢占资源，就会进入阻塞队列的链表中，当release释放state-1之后，新进来的线程会尝试占用资源，否则链表中的线程被唤醒并开始抢占资源(非公平锁)。
CyclicBarrier栅栏
CyclicBarrier的字面意思是可循环使用的屏障。他要做的事情是让一组线程到达一个屏障(也可以叫同步点)时被阻塞，直到最后一个线程到达屏障时，屏障才会打开，所有被屏障拦截的线程才会继续工作。
使用场景：当存在需要所有子任务都完成时，才执行主任务，这个时候就可以选择使用CyclicBarrier
注意：
- 如果因为某种原因(Exception)导致没有足够多的线程来调用await，这个时候会导致所有线程都会被阻塞
- 可以用await(timeout, unit)设置一个超时时间
- reset重置计数器，没有足够多的await就会抛出BrokenBarrierException
当执行完await后，所有线程执行完会执行CyclicBarrier构造器中的Runnable方法，并且会执行await之后的代码。
内部声明一个ReentrantLock和Condition以及Generation
demo
package com.elian.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* Describe:
*
* @author Elian
* @since 2022/3/14 1:03
*/
public class Horse implements Runnable {
private static int counter = 0;
private final int id = counter++;
private int strides = 0;
private static Random rand = new Random(47);
private static CyclicBarrier barrier;
public Horse(CyclicBarrier b) { barrier = b; }
@Override
public void run() {
try {
while(!Thread.interrupted()) {
synchronized(this) {
//赛马每次随机跑几步
strides += rand.nextInt(3);
}
barrier.await();
}
} catch(Exception e) {
e.printStackTrace();
}
}
public String tracks() {
StringBuilder s = new StringBuilder();
for(int i = 0; i < getStrides(); i++) {
s.append("*");
}
s.append(id);
return s.toString();
}
public synchronized int getStrides() { return strides; }
public String toString() { return "Horse " + id + " "; }
}
class HorseRace implements Runnable {
private static final int FINISH_LINE = 75;
private static List horses = new ArrayList();
private static ExecutorService exec = Executors.newCachedThreadPool();
@Override
public void run() {
StringBuilder s = new StringBuilder();
//打印赛道边界
for(int i = 0; i < FINISH_LINE; i++) {
s.append("=");
}
System.out.println(s);
//打印赛马轨迹
for(Horse horse : horses) {
System.out.println(horse.tracks());
}
//判断是否结束
for(Horse horse : horses) {
if(horse.getStrides() >= FINISH_LINE) {
System.out.println(horse + "won!");
exec.shutdownNow();
return;
}
}
//休息指定时间再到下一轮
try {
TimeUnit.MILLISECONDS.sleep(200);
} catch(InterruptedException e) {
System.out.println("barrier-action sleep interrupted");
}
}
public static void main(String[] args) {
CyclicBarrier barrier = new CyclicBarrier(7, new HorseRace());
for(int i = 0; i < 7; i++) {
Horse horse = new Horse(barrier);
horses.add(horse);
exec.execute(horse);
}
}
}
Condition
Condition是多线程协调通信的工具类，可以让某些线程一起等待某个条件(condition)，只有满足条件时，线程才会被唤醒。