4.共享模型之管程
4.1 共享带来的问题
Java的体现
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/6 13:00
*/
@Slf4j
public class Test14 {
static int i = 0;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(()->{
for(int j = 0; j < 5000; j++) {
i++;
}
});
Thread t2 = new Thread(()->{
for(int j = 0; j < 5000; j++) {
i--;
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(i);
}
}
结果
问题分析
以上的结果可能是正数、负数、零。为什么呢？因为 Java 中对静态变量的自增，自减并不是原子操作，要彻底理 解，必须从字节码来进行分析 例如对于 i++ 而言（i 为静态变量），实际会产生如下的 JVM 字节码指令：
getstatic i // 获取静态变量i的值
iconst_1 // 准备常量1
iadd // 自增
putstatic i // 将修改后的值存入静态变量i
而对应i–也是类似：
getstatic i // 获取静态变量i的值
iconst_1 // 准备常量1
isub // 自减
putstatic i // 将修改后的值存入静态变量i
而 Java 的内存模型如下，完成静态变量的自增，自减需要在主存和工作内存中进行数据交换：
如果是单线程以上 8 行代码是顺序执行（不会交错）没有问题：
但是多线程下可能会出现问题
临界区 Critical Section
-
一个程序运行多个线程本身是没有问题的
-
问题出在多个线程访问共享资源
-
- 多个线程读共享资源其实也没有问题
- 在多个线程对共享资源读写操作时发生指令交错，就会出现问题
-
一段代码块内如果存在对共享资源的多线程读写操作，称这段代码块为临界区
例如
static int counter = 0;
static void increment()
// 临界区
{
counter++;
}
static void decrement()
// 临界区
{
counter--;
}
竞态条件 Race Condition
多个线程在临界区内执行，由于代码的执行顺序序列不同而导致结果无法预测，称为发生了竞态条件
4.2 synchronized 解决方案
为了避免临界区的竞态条件发生，有多种手段可以达到目的
- 非阻塞的解决方案：synchronized，Lock
- 非阻塞式的解决方案：原子变量
synchronized
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/6 13:00
*/
@Slf4j
public class Test14 {
static int i = 0;
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(()-> {
for(int j = 0; j < 5000; j++) {
synchronized (Test14.class) {
i++;
}
}
});
Thread t2 = new Thread(()->{
for(int j = 0; j < 5000; j++) {
synchronized (Test14.class) {
i--;
}
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(i);
}
}
图分析
思考
synchronized实际是用对象锁保证了临界区内代码的原子性，临界区内的代码对外是不可分割的，不会被线程切换所打断。
- 如果把 synchronized(obj) 放在 for 循环的外面，如何理解？-- 原子性 如果 t1
- synchronized(obj1) 而 t2 synchronized(obj2) 会怎样运作？-- 锁对象 如果 t1
- synchronized(obj) 而 t2 没有加会怎么样？如何理解？-- 锁对象
面向对象改进
把需要保护的变量放入一个类中
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/6 13:00
*/
@Slf4j
public class Test14 {
static Object o1 = new Object();
static Object o2 = new Object();
static int i = 0;
public static void main(String[] args) throws InterruptedException {
Room room = new Room();
Thread t1 = new Thread(()-> {
for(int j = 0; j < 5000; j++) {
room.increment();
}
});
Thread t2 = new Thread(()->{
for(int j = 0; j < 5000; j++) {
room.decrement();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(room.getValue());
}
}
class Room{
int value = 0;
public void increment(){
synchronized (this) {
value++;
}
}
public void decrement(){
synchronized (this) {
value--;
}
}
public int getValue(){
synchronized (this) {
return value;
}
}
}
4.3 方法上的synchronized
锁住的是this
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/6 13:00
*/
@Slf4j
public class Test14 {
static Object o1 = new Object();
static Object o2 = new Object();
static int i = 0;
public static void main(String[] args) throws InterruptedException {
Room room = new Room();
Thread t1 = new Thread(()-> {
for(int j = 0; j < 5000; j++) {
room.increment();
}
});
Thread t2 = new Thread(()->{
for(int j = 0; j < 5000; j++) {
room.decrement();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(room.getValue());
}
}
class Room{
int value = 0;
public synchronized void increment(){
value++;
}
public synchronized void decrement(){
value--;
}
public synchronized int getValue(){
return value;
}
}
加在静态方法上锁的是类对象O.class
不加synchronized的方法
不加synchronized的方法就好比不遵守规则的人，不老实排队
“线程八锁”
情况1：12或21
@Slf4j(topic = "c.Number")
class Number{
public synchronized void a() {
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n1.b(); }).start();
}
情况2： 1s后12，或 2 1s后 1
sleep是不会释放锁
@Slf4j(topic = "c.Number")
class Number{
public synchronized void a() {
sleep(1);
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n1.b(); }).start();
}
情况3： 3 1s 12 或 23 1s 1 或 32 1s 1
class Number{
public synchronized void a() {
sleep(1);
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
public void c() {
log.debug("3");
}
}
public static void main(String[] args) {
Number n1 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n1.b(); }).start();
new Thread(()->{ n1.c(); }).start();
}
情况4： 2 1s 后 1
@Slf4j(topic = "c.Number")
class Number{
public synchronized void a() {
sleep(1);
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
Number n2 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n2.b(); }).start();
}
情况5： 2 1s 后 1
@Slf4j(topic = "c.Number")
class Number{
public static synchronized void a() {
sleep(1);
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n1.b(); }).start();
}
情况6： 1s 后12， 或 2 1s后 1
@Slf4j(topic = "c.Number")
class Number{
public static synchronized void a() {
sleep(1);
log.debug("1");
}
public static synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n1.b(); }).start();
}
情况7： 2 1s 后 1
@Slf4j(topic = "c.Number")
class Number{
public static synchronized void a() {
sleep(1);
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
Number n2 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n2.b(); }).start();
}
情况8： 1s 后12， 或 2 1s后 1
class Number{
public static synchronized void a() {
sleep(1);
log.debug("1");
}
public static synchronized void b() {
log.debug("2");
}
}
public static void main(String[] args) {
Number n1 = new Number();
Number n2 = new Number();
new Thread(()->{ n1.a(); }).start();
new Thread(()->{ n2.b(); }).start();
}
4.4 变量的线程安全分析
成员变量和静态变量是否线程安全？
-
如果他们没有共享，则线程完全
-
如果它们被共享了，根据它们的状态是否能够改变，又分两种情况
-
- 如果只有读操作，线程安全
- 如果有读写操作，则这段代码块属于临界区，需要考虑线程安全
局部变量是否线程安全？
-
局部变量是线程安全
-
但局部变量引用的对象未必
-
- 如果该对象没有逃离方法的作用范围，它是线程安全的
- 如果该对象逃离方法的作用范围，需要考虑线程安全
局部变量线程安全分析
public static void test1(){
int i = 10;
i++;
}
每个线程调用test1()方法时局部变量i，会在每个线程的栈帧内存中被创建多份，因此不存在共享
反编译文件
public static void test1();
descriptor: ()V
flags: ACC_PUBLIC, ACC_STATIC
Code:
stack=1, locals=1, args_size=0
0: bipush 10
2: istore_0
3: iinc 0, 1
6: return
LineNumberTable:
line 10: 0
line 11: 3
line 12: 6
LocalVariableTable:
Start Length Slot Name Signature
3 4 0 i I
常见线程安全类
- String
- Integer
- StringBuffer
- Random
- Vector
- Hashtable
- Java.util.concurrent包下的所有类
多个线程调用它们同一个实例的某个方法时，是线程安全的
- 它们的每个方法是原子的
- 但注意它们多个方法的组合不是原子的
不可变线程安全性
String、Integer 等都是不可变类，因为其内部的状态不可以改变，因此它们的方法都是线程安全的
4.5 Monitor概念
Java对象头
以32位虚拟机为例
普通对象
数组对象
其中Mark Word结构
64位虚拟机Mark Word
Monitor
Monitor被翻译为监视器或管程
每个Java对象都可以关联一个Monitor对象，如果使用synchronized给对象上锁之后，该对象头的Mark Work中就被设置指向Monitor对象指针。
- 刚开始Monitor中Owner为null
- 当线程1进入临界区，需要拿到obj的锁，就会根据obj对象的Mark Word找到对应的Monitor，看Owner是否为null
- 如果为null，则将Owner指向线程1，表示线程1拿到锁
- 当线程2进入临界区的时候，发现Owner不为null，就会被添加到EntryList中，变成BLOCKED状态
- 等待线程1释放锁，也就是使该Owner为null时看EntryList中的哪个线程会拿到锁。
synchronized
对应的字节码文件为
4.6 synchronized原理进阶
1.1. 轻量级锁
轻量级锁的使用场景：如果一个对象虽然有多线程访问，当多线程访问的时间是错开的（也就是没有竞争），那么可以使用轻量级锁来优化。
轻量级锁对使用者是透明的，即语法仍然是synchronized
假设有两个方法同步块，利用同一个对象加锁
static final Object obj = new Object();
public static void method1(){
synchronized(obj){
// 同步块a
method2();
}
}
public static void method2(){
synchronized(obj){
// 同步块b
}
}
- 创建锁记录（Lock Record）对象，每个线程的栈帧都会包含一个锁记录的结构，内部可以存储锁定对象的Mark Word
- 让锁记录中Object reference指向锁对象，并尝试用cas替换Object的Mark Word，将Mark Word的值存入锁记录
- 如果cas替换成功，对象头中存储了锁记录地址和状态00，表示由该线程给对象加锁
-
如果cas失败，有两种情况：
-
- 如果是其它线程已经持有了改object的轻量级锁，这时表明有竞争，进入锁膨胀过程
- 如果是自己执行了synchronized锁重入，那么再添加一条Lock Record作为重入的计数
- 当退出synchronized代码块（解锁时）如果有取值为null的锁记录，表示有重入，这时重置锁记录，表示重入计数减一
-
当退出synchronized代码块（解锁时）锁记录的值不为null，这时使用cas将Mark Word的值恢复给对象头
-
- 成功，说明解锁成功
- 失败，说明轻量级锁进行了锁膨胀或已经升级为重量级锁，进入重量级锁的解锁流程
1.2. 锁膨胀
如果在尝试加轻量级锁的过程中，CAS操作无法成功，这时一种情况就是有其它线程为此对象加上了轻量级锁（有竞争），这时需要进行锁膨胀，将轻量级锁变为重量级锁
static Object obj = new Object();
public static void method1(){
synchornized(obj){
// 同步块
}
}
- 当Thread-1进行轻量级加锁时，Thread-0已经对该对象加了轻量级锁
-
这时Thread-1加轻量级锁失败，进入锁膨胀流程
-
- 即为Object对象申请Monitor锁，让Object指向重量级锁的地址
- 然后自己进入Monitor的EntryList BlOCKED
- 当Thread-0退出同步代码块时，使用cas将Mark Word的值恢复给对象头，失败。这是会进入重量级解锁流程，即按照Monitor地址找到Monitor对象，设置Owner为null，唤醒EntryList中BLOCKED线程
1.3. 自旋优化
重量级锁竞争的时候，还可以使用自旋来进行优化，如果当前线程自旋成功（即这时候持锁线程已经退出了同步块，释放了锁），这时当前线程就可以避免阻塞
自旋重试成功的情况：
自旋重试失败的情况：
- 在Java6之后自旋锁是自适应的，比如对象刚刚的一次自选操作成功过，那么认为这次自旋成功的可能性比较高，会多尝试几次，反之，则少尝试几次
- 自旋会占用cpu时间，单核cpu自旋就是浪费时间，多核cpu才能发挥出性能优势
- Java7以后不能控制是否开启自旋功能
1.4. 偏向锁
轻量级锁在没有竞争时（就自己这个线程），每次重入仍然需要执行CAS操作
Java6中引入了偏向锁来做进一步优化：只有第一次使用CAS将线程ID设置到对象的Mark Word头，之后发现这个线程ID是自己的就表示没有竞争，不用重新CAS。以后只要不发生竞争，这个对象就归该线程所有。
列如
static Object obj = new Object();
public static void method1(){
synchornized(obj){
// 同步块
method2();
}
}
public static void method2(){
synchornized(obj){
// 同步块
method3();
}
}
public static void method3(){
synchornized(obj){
// 同步块
}
}
图解
1.4.1. 偏向状态
回忆以下对象头格式
一个对象创建时：
- 如果开启了偏向锁（默认开启），那么对象创建后，markword值为0x05即最后3位101，这时它的thread、epoch、age都为0
- 偏向锁默认是延迟的，不会在程序启动时立即生效，如果想避免延迟，可以加VM参数 -XX:BiasedLockingStartupDelay=0来禁用延迟
- 如果没有开启偏向锁，那么对象创建后，markword值位0x01即最后3位001，这时它的hashcode，age都为0，第一次用到hashcode时才会被赋值
1）测试延迟特性
2）测试偏向锁
3)测试禁用
4）测试hashCode
1.4.2. 撤销-调用对象hashCode
调用了对象的hashCode，但偏向锁的对象MarkWord中存储的是线程id，如果调用hashCode会导致偏向锁被撤销
- 轻量级锁会在锁记录中记录hashCode
- 重量级锁会在Monitor中记录hashCode
1.4.3. 撤销-其它线程使用对象
当有其它线程使用偏向锁对象时，会将偏向锁升级为轻量级锁
1.4.4. 撤销-调用wait/notify
1.4.5. 批量重偏向
如果对象虽然被多个线程访问，但没有竞争，这时偏向了线程T1的对象仍有重新偏向T2，重偏向会重置对象的ThreadID
当撤销偏向锁阈值超过20次后，jvm会这样觉得，我是不是偏向错了呢，于是会在给这些对象加锁时重新偏向至加锁过程
1.4.6. 批量撤销
当撤销偏向锁阈值超过40次后，jvm会这样觉得，我自己确实偏向错了，根本就不该偏向。于是整个类的所有对象都会变为不可偏向的，新建的对象也是不可偏向的。
1.4.7. 锁消除
4.7 原理之wait/notify
- Owner线程发现条件不满足，调用wait方法，即可进入WaitSet变为WAITING状态
- BLOCKED和WAITING线程都处于阻塞状态，不占用CPU时间片
- BLOCKED线程会在Owner线程释放锁时唤醒
- WAITING线程会在Owner线程调用notify或notifyAll时唤醒，但唤醒后并不意味着立即获得锁，人需要进入EntryList重新竞争
API介绍
obj.wait()让进入object监视器的线程到waitSet等待obj.wait(long timeout)最多等待timeout毫秒obj.notify()在object上正在waitSet等待的线程中挑一个唤醒obj.notifyAll()让object上正在waitSet等待的线程全部唤醒
它们都是线程之间进行协作的手段，都属于Object对象的方法。必须获得此对象的锁，才能调用这几个方法。否则会抛异常。
4.8 wait notify的正确姿势
sleep(long n) 和 wait(long n)的区别
- sleep是Thread方法，而wait是Object的方法
- sleep不需要强制更synchronized配合使用，但wait需要跟synchronized一起使用
- sleep在睡眠的同时不会释放锁，但wait在等待的时候会释放锁
- 它们状态TIMED_WAITING
Step1
import lombok.extern.slf4j.Slf4j;
import static java.lang.Thread.sleep;
/**
* @author xc
* @date 2023/5/7 23:11
*/
@Slf4j
public class Test17 {
static final Object room = new Object();
static boolean hasCigarette = false;
static boolean hasTakeout = false;
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
synchronized (room) {
log.debug("有烟没？[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟，先歇会！");
try {
sleep(2);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
log.debug("有烟没？[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
}
}
}, "小南").start();
for (int i = 0; i < 5; i++) {
new Thread(() -> {
synchronized (room) {
log.debug("可以开始干活了");
}
}, "其它人").start();
}
sleep(1);
new Thread(() -> {
// 这里能不能加 synchronized (room)？
hasCigarette = true;
log.debug("烟到了噢！");
}, "送烟的").start();
}
}
输出
- 其它干活的线程都要一直阻塞，效率太低
- 小南线程必须睡足2s后才能醒来，就算烟提前送到，也无法立即醒来
- 加了synchronized(room)后，就好比小南在里面反锁门睡觉，烟根本没法送进门，main没加synchronized就好比main线程是翻窗户进来的
- 解决方法，使用wait-notify机制
Step2
import lombok.extern.slf4j.Slf4j;
import static java.lang.Thread.sleep;
/**
* @author xc
* @date 2023/5/7 23:11
*/
@Slf4j
public class Test17 {
static final Object room = new Object();
static boolean hasCigarette = false;
static boolean hasTakeout = false;
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
synchronized (room) {
log.debug("有烟没？[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
log.debug("有烟没？[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
}
}
}, "小南").start();
for (int i = 0; i < 5; i++) {
new Thread(() -> {
synchronized (room) {
log.debug("可以开始干活了");
}
}, "其它人").start();
}
sleep(1);
new Thread(() -> {
// 这里能不能加 synchronized (room)？
synchronized (room) {
hasCigarette = true;
log.debug("烟到了噢！");
room.notify();
}
}, "送烟的").start();
}
}
结果
- 解决了其它干活的线程阻塞的问题
- 但如果由其它线程也在等待条件呢？
Step3
import lombok.extern.slf4j.Slf4j;
import static java.lang.Thread.sleep;
/**
* @author xc
* @date 2023/5/7 23:11
*/
@Slf4j
public class Test17 {
static final Object room = new Object();
static boolean hasCigarette = false;
static boolean hasTakeout = false;
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
synchronized (room) {
log.debug("有烟没？[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("有烟没？[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小南").start();
new Thread(() -> {
synchronized (room) {
Thread thread = Thread.currentThread();
log.debug("外卖送到没？[{}]", hasTakeout);
if (!hasTakeout) {
log.debug("没外卖，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("外卖送到没？[{}]", hasTakeout);
if (hasTakeout) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小女").start();
sleep(1);
new Thread(() -> {
synchronized (room) {
hasTakeout = true;
log.debug("外卖到了噢！");
room.notify();
}
}, "送外卖的").start();
}
}
结果
Step4
import lombok.extern.slf4j.Slf4j;
import static java.lang.Thread.sleep;
/**
* @author xc
* @date 2023/5/7 23:11
*/
@Slf4j
public class Test17 {
static final Object room = new Object();
static boolean hasCigarette = false;
static boolean hasTakeout = false;
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
synchronized (room) {
log.debug("有烟没？[{}]", hasCigarette);
if (!hasCigarette) {
log.debug("没烟，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("有烟没？[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小南").start();
new Thread(() -> {
synchronized (room) {
Thread thread = Thread.currentThread();
log.debug("外卖送到没？[{}]", hasTakeout);
if (!hasTakeout) {
log.debug("没外卖，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("外卖送到没？[{}]", hasTakeout);
if (hasTakeout) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小女").start();
sleep(1);
new Thread(() -> {
synchronized (room) {
hasTakeout = true;
log.debug("外卖到了噢！");
room.notifyAll();
}
}, "送外卖的").start();
}
}
结果
- 用notifyAll仅解决某个线程的唤醒问题，但使用if+wait判断仅有一次机会，一但条件不成立，就没有重新判断的机会了
- 解决方法，用while+wait，但条件不成立，再次wait
Step5
import lombok.extern.slf4j.Slf4j;
import static java.lang.Thread.sleep;
/**
* @author xc
* @date 2023/5/7 23:11
*/
@Slf4j
public class Test17 {
static final Object room = new Object();
static boolean hasCigarette = false;
static boolean hasTakeout = false;
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
synchronized (room) {
log.debug("有烟没？[{}]", hasCigarette);
while (!hasCigarette) {
log.debug("没烟，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("有烟没？[{}]", hasCigarette);
if (hasCigarette) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小南").start();
new Thread(() -> {
synchronized (room) {
Thread thread = Thread.currentThread();
log.debug("外卖送到没？[{}]", hasTakeout);
if (!hasTakeout) {
log.debug("没外卖，先歇会！");
try {
room.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("外卖送到没？[{}]", hasTakeout);
if (hasTakeout) {
log.debug("可以开始干活了");
} else {
log.debug("没干成活...");
}
}
}, "小女").start();
sleep(1);
new Thread(() -> {
synchronized (room) {
hasTakeout = true;
log.debug("外卖到了噢！");
room.notifyAll();
}
}, "送外卖的").start();
}
}
结果
使用套路
synchronized(lock){
while(条件不成立){
lock.wait();
}
// 干活
}
// 另一个线程
synchronized(lock){
lock.notifyAll();
}
同步模式之保护性暂停
1. 定义
即Guarded Suspension，用在一个线程等待另一个线程的执行结果
要点：
- 有一个结果需要从一个线程传递到另一个线程，让它们关联共一个GuardedObject
- 如果有结果不断从一个线程到另一个线程那么可以使用消息队列
- JDK中，join的实现，Future的实现，采用的就是此模式
- 因为要等待另一方的结果，因此归类到同步模式
2. 实现
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.List;
/**
* @author xc
* @date 2023/5/8 13:08
*/
@Slf4j
class Test20{
public static void main(String[] args) {
GuardedObject guardedObject = new GuardedObject();
new Thread(() -> {
try {
// 子线程执行下载
List<String> response = download();
log.debug("download complete...");
guardedObject.complete(response);
} catch (IOException e) {
e.printStackTrace();
}
}).start();
log.debug("waiting...");
// 主线程阻塞等待
Object response = guardedObject.get();
log.debug("get response: [{}] lines", ((List<String>) response).size());
}
}
class GuardedObject {
private Object response;
private final Object lock = new Object();
public Object get() {
synchronized (lock) {
// 条件不满足则等待
while (response == null) {
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
return response;
}
}
public void complete(Object response) {
synchronized (lock) {
// 条件满足，通知等待线程
this.response = response;
lock.notifyAll();
}
}
}
3. 带超时版GuardedObject
如果要控制超时时间
class GuardedObjectV2 {
private Object response;
private final Object lock = new Object();
public Object get(long millis) {
synchronized (lock) {
// 1) 记录最初时间
long begin = System.currentTimeMillis();
// 2) 已经经历的时间
long timePassed = 0;
while (response == null) {
// 4) 假设 millis 是 1000，结果在 400 时唤醒了，那么还有 600 要等
long waitTime = millis - timePassed;
log.debug("waitTime: {}", waitTime);
if (waitTime <= 0) {
log.debug("break...");
break;
}
try {
lock.wait(waitTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 3) 如果提前被唤醒，这时已经经历的时间假设为 400
timePassed = System.currentTimeMillis() - begin;
log.debug("timePassed: {}, object is null {}",
timePassed, response == null);
}
return response;
}
}
public void complete(Object response) {
synchronized (lock) {
// 条件满足，通知等待线程
this.response = response;
log.debug("notify...");
lock.notifyAll();
}
}
}
异步模式之生产者/消费者
定义
要点
- 与前面的保护性暂停中的GuardObject不同，不需要产生结果和消费结果的线程一一对应
- 消费队列可以用来平衡生产和消费的线程资源
- 生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果
- 消息队列是有容量限制的，满时不会再加入数据，空时不会在小号数据
- JDK中各种阻塞队列，采用的就是这种模式
4.9 Park & Unpark
基本使用
它们是LockSupport类中的方法
// 暂停当前线程
LockSupport.park();
// 恢复某个线程的运行
LockSupport.unpark(暂停线程对象)
先park再unpark
Thread t1 = new Thread(() -> {
log.debug("start...");
sleep(1);
log.debug("park...");
LockSupport.park();
log.debug("resume...");
},"t1");
t1.start();
sleep(2);
log.debug("unpark...");
LockSupport.unpark(t1);
输出
18:42:52.585 c.TestParkUnpark [t1] - start...
18:42:53.589 c.TestParkUnpark [t1] - park...
18:42:54.583 c.TestParkUnpark [main] - unpark...
18:42:54.583 c.TestParkUnpark [t1] - resume...
特点
与Object的wait & notify相比
- wait，notify和notifyAll必须配合Object Monitor一起使用，而park，unpark不必
- park & unpark是以线程为单位来【阻塞】和【唤醒】线程，而notify只能随机唤醒一个等待线程，notifyAll是唤醒所有等待线程，就不那么【精确】
- park & unpark可以先unpark，而wait & notify不能先notify
原理之park & unpark
- 当前线程调用Unsafe.park()方法
- 检查_counter,本情况为0，这时，获得_mutex互斥锁
- 线程进入_cond条件变量阻塞
- 设置_counter=0
- 调用Unsafe.unpark(Thread_0)方法，设置_counter为1
- 当前线程调用Unsafe.park()方法
- 检查_counter，本情况为1，这时线程无需阻塞，继续运行
- 设置_counter为0
4.10 重新理解线程状态转换
情况1 NEW --> RUNNABLE
- 当调用t.start()方法时，由NEW --> RUNNABLE
情况2 RUNNABLE <–> WAITING
t线程用synchronized(obj)获取了对象锁后
-
调用obj.wait()方法时，t线程从RUNNABLE --> WAITING
-
调用obj.notify()，obj.notifyAll()，obj.interrupt()方法时，
-
- 竞争锁成功，t线程从WAITING --> RUNNABLE
- 竞争锁失败，t线程从WAITING --> BLOCKED
情况3 RUNNABLE <–> WAITING
-
当前线程调用t.join()方法时，当前线程从RUNNABLE --> WAITING
-
- 注意是当前线程在t线程对象的监视器上等待
-
t线程运行结束，或调用了当前线程的interrupt()时，当前线程从WAITING --> RUNNABLE
情况4 RUNNABLE <–> WAITING
- 当前线程调用LockSupport.park()方法会让当前线程从RUNNABLE --> WAITING
- 调用LockSupport.unpark(目标线程)或调用了线程的interrupt()，会让目标线程从WAITING --> RUNNABLE
情况5 RUNNABLE <–> TIMED_WAITING
t线程调用synchronized(obj)对象锁后
-
调用obj.wait(long n)方法时，t线程从RUNNABLE --> TIMED_WAITING
-
t线程等待时间超过了n毫秒，或者调用obj.notify()，obj.notifyAll()，t.interrupt()时
-
- 竞争锁成功，t线程从TIMED_WAITING --> RUNNABLE
- 竞争锁失败，t线程从TIMED_WAITING --> BLOCKED
情况6 RUNNABLE --> TIMED_WAITING
-
当前线程调用t.join(long n)方法时，当前线程从RUNNABLE --> TIMED_WAITING
-
- 注意是当前线程在t线程对象的监视器上等待
-
当前线程等待时间超过了n毫秒，或t线程运行结束，或调用了当前线程的interrupt()时，当前线程从TIMED_WAITING --> RUNNABLE
情况7 RUNNABLE --> TIMED_WAITING
- 当前线程调用Thread.sleep(long n)，当前线程从RUNNABLE --> TIMED_WAITING
- 当前线程等待时间超过了n毫秒，当前线程从TIMED_WAITING --> RUNNABLE
情况8 RUNNABLE --> TIMED_WAITING
- 当前线程调用LockSupport.parkNanos(long nanos)或LockSupport.parkUntil(long millis)时，当前线程从RUNNABLE --> TIMED_WAITING
- 调用LockSupport.unpark(目标线程)或调用了线程的interrupt()，或是等待超时，会让目标线程从TIMED_WAITING–>RUNNABLE
情况9 RUNNABLE <–> BLOCKED
- t线程用synchronized(obj)获取对象锁时如果竞争失败，从RUNNABLE–> BLOCKED
- 持obj锁线程的同步代码块执行完毕，会唤醒该对象上所有BLOCKED线程重新竞争，如果其中t线程竞争成功，从BLOCKED–> RUNNABLE，其它失败的线程仍然BLOCKED
情况10 RUNNABLE <–> TERMINATED
当前线程所有代码运行完毕，进入TERMINATED
4.11 多把锁
多把不相干的锁
一间大屋子有两个功能：睡觉、学习，互不相干。 现在小南要学习，小女要睡觉，但如果只用一间屋子（一个对象锁）的话，那么并发度很低 解决方法是准备多个房间（多个对象锁）
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.List;
/**
* @author xc
* @date 2023/5/8 13:08
*/
@Slf4j
public class Test20{
public static void main(String[] args) throws InterruptedException {
BigRoom bigRoom = new BigRoom();
new Thread(()->{
try {
bigRoom.sleep();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
},"小南").start();
new Thread(()->{
try {
bigRoom.study();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
},"小女").start();
}
}
@Slf4j
class BigRoom {
public void sleep() throws InterruptedException {
synchronized (this) {
log.debug("sleeping 2 小时");
Thread.sleep(2000);
}
}
public void study() throws InterruptedException {
synchronized (this) {
log.debug("study 1 小时");
Thread.sleep(1000);
}
}
}
结果
改进
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.List;
/**
* @author xc
* @date 2023/5/8 13:08
*/
@Slf4j
public class Test20{
public static void main(String[] args) throws InterruptedException {
BigRoom bigRoom = new BigRoom();
new Thread(()->{
try {
bigRoom.sleep();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
},"小南").start();
new Thread(()->{
try {
bigRoom.study();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
},"小女").start();
}
}
@Slf4j
class BigRoom {
// 多把锁互不影响
private Object sleepLock = new Object();
private Object studyLock = new Object();
public void sleep() throws InterruptedException {
synchronized (sleepLock) {
log.debug("sleeping 2 小时");
Thread.sleep(2000);
}
}
public void study() throws InterruptedException {
synchronized (studyLock) {
log.debug("study 1 小时");
Thread.sleep(1000);
}
}
}
结果
4.12 活跃性
死锁
一个线程需要同时获取多把锁，这时就容易发生死锁
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/8 16:38
*/
@Slf4j
public class Test21 {
private static Object a = new Object();
private static Object b = new Object();
public static void main(String[] args) {
new Thread(()->{
synchronized (a) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
synchronized (b) {
log.debug("a操作");
}
}
}).start();
new Thread(()->{
synchronized (b) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
synchronized (a) {
log.debug("b操作");
}
}
}).start();
}
}
结果
一直不打印操作
定位死锁
- 检测死锁可以使用jconsole工具，或者使用jps定位进程id，再用jstack定位死锁
哲学家就餐问题
- 有五位哲学家，围坐在圆桌旁。
- 他们只做两件事，思考和吃饭，思考一会吃口饭，吃完饭后接着思考。
- 吃饭时要用两根筷子吃，桌上共有 5 根筷子，每位哲学家左右手边各有一根筷子。
- 如果筷子被身边的人拿着，自己就得等待
活锁
活锁出现在两个线程互相改变对方的结束条件，最后谁也无法结束，例如
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/8 16:38
*/
@Slf4j
public class Test21 {
static volatile int count = 10;
private final Object lock = new Object();
public static void main(String[] args) {
new Thread(()->{
while (count > 0) {
try {
Thread.sleep(2);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
count--;
log.debug(String.valueOf(count));
}
}).start();
new Thread(()->{
while (count < 20) {
try {
Thread.sleep(2);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
count++;
log.debug(String.valueOf(count));
}
}).start();
}
}
饥饿
4.13 ReentrantLock
相对于synchronized它具备如下特点
- 可中断
- 可以设置超时时间
- 可以设置为公平锁
- 支持多个变量条件
与synchronized一样，都可以重入
基本语法
// 获取锁
reentrantLock.lock();
try {
// 临界区
} finally {
// 释放锁
reentrantLock.unlock();
}
可重入
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author xc
* @date 2023/5/8 17:12
*/
@Slf4j
public class Test22 {
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {
lock.lock();
try {
log.debug("进入m1");
m1();
} finally {
lock.unlock();
}
}
public static void m1(){
lock.lock();
try {
log.debug("进入m2");
m2();
} finally {
lock.unlock();
}
}
public static void m2(){
lock.lock();
try {
log.debug("进入m3");
} finally {
lock.unlock();
}
}
}
可打断
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author xc
* @date 2023/5/8 17:29
*/
@Slf4j
public class Test23 {
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(() -> {
log.debug("启动...");
try {
lock.lockInterruptibly();
} catch (InterruptedException e) {
e.printStackTrace();
log.debug("等锁的过程中被打断");
return;
}
try {
log.debug("获得了锁");
} finally {
lock.unlock();
}
}, "t1");
lock.lock();
log.debug("获得了锁");
t1.start();
try {
Thread.sleep(1000);
t1.interrupt();
log.debug("执行打断");
} finally {
lock.unlock();
}
}
}
结果
锁超时
超时失败
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author xc
* @date 2023/5/8 18:15
*/
@Slf4j
public class Test24 {
private static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(()->{
try {
if (!lock.tryLock(2, TimeUnit.SECONDS)) {
log.debug("获取不到锁");
return;
}
} catch (InterruptedException e) {
e.printStackTrace();
log.debug("获取不到共享锁");
return;
}
try {
log.debug("获取到锁");
} finally {
lock.unlock();
}
},"t1");
lock.lock();
log.debug("获取到锁");
t1.start();
Thread.sleep(2000);
lock.unlock();
log.debug("释放锁");
}
}
公平锁
公平锁一般没有必要，会降低并发度。
条件变量
synchronized中也有条件变量，就是我们讲原理时那个waitSet休息室，当条件不满足时进入waitSet等待ReentrantLock的条件变量比synchronized强大之处在于它是支持多个条件变量的，这就好比
- synchronized是那些不满足条件的线程都在一间休息室等消息
- 而ReentrantLock支持多间休息室，有专门等烟的休息室、专门等早餐的休息室，唤醒时也是按休息室来唤醒
使用要点：
- await前需要获取锁
- await执行后，会释放锁，进入conditionObject等待
- await的线程被唤醒（打断、或超时）取重新竞争lock锁
- 竞争lock锁成功后，从await后继续执行
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author xc
* @date 2023/5/8 18:52
*/
@Slf4j
public class Test25 {
static ReentrantLock lock = new ReentrantLock();
static Condition waitCigaretteQueue = lock.newCondition();
static Condition waitbreakfastQueue = lock.newCondition();
static volatile boolean hasCigrette = false;
static volatile boolean hasBreakfast = false;
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
try {
lock.lock();
while (!hasCigrette) {
try {
waitCigaretteQueue.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("等到了它的烟");
} finally {
lock.unlock();
}
}).start();
new Thread(() -> {
try {
lock.lock();
while (!hasBreakfast) {
try {
waitbreakfastQueue.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
log.debug("等到了它的早餐");
} finally {
lock.unlock();
}
}).start();
Thread.sleep(1000);
sendBreakfast();
Thread.sleep(1000);
sendCigarette();
}
private static void sendCigarette() {
lock.lock();
try {
log.debug("送烟来了");
hasCigrette = true;
waitCigaretteQueue.signal();
} finally {
lock.unlock();
}
}
private static void sendBreakfast() {
lock.lock();
try {
log.debug("送早餐来了");
hasBreakfast = true;
waitbreakfastQueue.signal();
} finally {
lock.unlock();
}
}
}
同步模式之顺序控制
固定运行顺序
先打印2，再打印1
wait & notify
import lombok.extern.slf4j.Slf4j;
/**
* @author xc
* @date 2023/5/8 19:05
*/
@Slf4j
public class Test26 {
private static Object o = new Object();
private static boolean isPrint = false;
public static void main(String[] args) {
Thread t1 = new Thread(()->{
synchronized (o){
while (!isPrint) {
try {
o.wait();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
log.debug("1");
}
});
Thread t2 = new Thread(()->{
synchronized (o){
log.debug("2");
isPrint = true;
o.notify();
}
});
t1.start();
t2.start();
}
}
park & unPark
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.LockSupport;
/**
* @author xc
* @date 2023/5/8 19:05
*/
@Slf4j
public class Test26 {
private static Object o = new Object();
private static boolean isPrint = false;
public static void main(String[] args) {
Thread t1 = new Thread(()->{
LockSupport.park();
log.debug("1");
});
Thread t2 = new Thread(()->{
log.debug("2");
isPrint = true;
LockSupport.unpark(t1);
});
t1.start();
t2.start();
}
}
交替输出
线程1输出a 5次，线程2输出b 5次，线程3输出c 5次。现在要求输出abcabcabcabbcabc怎么实现
wait & notify
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.LockSupport;
/**
* @author xc
* @date 2023/5/8 19:05
*/
@Slf4j
public class Test26 {
public static void main(String[] args) {
SyncWaitNotify syncWaitNotify = new SyncWaitNotify(1, 5);
new Thread(() -> {
syncWaitNotify.print(1, 2, "a");
}).start();
new Thread(() -> {
syncWaitNotify.print(2, 3, "b");
}).start();
new Thread(() -> {
syncWaitNotify.print(3, 1, "c");
}).start();
}
}
class SyncWaitNotify {
private int flag;
private int loopNumber;
public SyncWaitNotify(int flag, int loopNumber) {
this.flag = flag;
this.loopNumber = loopNumber;
}
public void print(int waitFlag, int nextFlag, String str) {
for (int i = 0; i < loopNumber; i++) {
synchronized (this) {
while (this.flag != waitFlag) {
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.print(str);
flag = nextFlag;
this.notifyAll();
}
}
}
}
Lock条件变量版
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author xc
* @date 2023/5/8 19:57
*/
@Slf4j
public class Test27 {
public static void main(String[] args) {
AwaitSignal as = new AwaitSignal(5);
Condition aWaitSet = as.newCondition();
Condition bWaitSet = as.newCondition();
Condition cWaitSet = as.newCondition();
new Thread(() -> {
as.print("a", aWaitSet, bWaitSet);
}).start();
new Thread(() -> {
as.print("b", bWaitSet, cWaitSet);
}).start();
new Thread(() -> {
as.print("c", cWaitSet, aWaitSet);
}).start();
as.start(aWaitSet);
}
}
@Slf4j
class AwaitSignal extends ReentrantLock {
public void start(Condition first) {
this.lock();
try {
log.debug("start");
first.signal();
} finally {
this.unlock();
}
}
public void print(String str, Condition current, Condition next) {
for (int i = 0; i < loopNumber; i++) {
this.lock();
try {
current.await();
log.debug(str);
next.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
this.unlock();
}
}
}
// 循环次数
private int loopNumber;
public AwaitSignal(int loopNumber) {
this.loopNumber = loopNumber;
}
}
Park Unpark版
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.LockSupport;
/**
* @author xc
* @date 2023/5/8 20:00
*/
@Slf4j
public class Test28 {
public static void main(String[] args) {
SyncPark syncPark = new SyncPark(5);
Thread t1 = new Thread(() -> {
syncPark.print("a");
});
Thread t2 = new Thread(() -> {
syncPark.print("b");
});
Thread t3 = new Thread(() -> {
syncPark.print("c\n");
});
syncPark.setThreads(t1, t2, t3);
syncPark.start();
}
}
class SyncPark {
private int loopNumber;
private Thread[] threads;
public SyncPark(int loopNumber) {
this.loopNumber = loopNumber;
}
public void setThreads(Thread... threads) {
this.threads = threads;
}
public void print(String str) {
for (int i = 0; i < loopNumber; i++) {
LockSupport.park();
System.out.print(str);
LockSupport.unpark(nextThread());
}
}
private Thread nextThread() {
Thread current = Thread.currentThread();
int index = 0;
for (int i = 0; i < threads.length; i++) {
if(threads[i] == current) {
index = i;
break;
}
}
if(index < threads.length - 1) {
return threads[index+1];
} else {
return threads[0];
}
}
public void start() {
for (Thread thread : threads) {
thread.start();
}
LockSupport.unpark(threads[0]);
}
}
this.lock();
try {
current.await();
log.debug(str);
next.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
this.unlock();
}
}
}
// 循环次数
private int loopNumber;
public AwaitSignal(int loopNumber) {
this.loopNumber = loopNumber;
}
}
Park Unpark版
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.LockSupport;
/**
* @author xc
* @date 2023/5/8 20:00
*/
@Slf4j
public class Test28 {
public static void main(String[] args) {
SyncPark syncPark = new SyncPark(5);
Thread t1 = new Thread(() -> {
syncPark.print("a");
});
Thread t2 = new Thread(() -> {
syncPark.print("b");
});
Thread t3 = new Thread(() -> {
syncPark.print("c\n");
});
syncPark.setThreads(t1, t2, t3);
syncPark.start();
}
}
class SyncPark {
private int loopNumber;
private Thread[] threads;
public SyncPark(int loopNumber) {
this.loopNumber = loopNumber;
}
public void setThreads(Thread... threads) {
this.threads = threads;
}
public void print(String str) {
for (int i = 0; i < loopNumber; i++) {
LockSupport.park();
System.out.print(str);
LockSupport.unpark(nextThread());
}
}
private Thread nextThread() {
Thread current = Thread.currentThread();
int index = 0;
for (int i = 0; i < threads.length; i++) {
if(threads[i] == current) {
index = i;
break;
}
}
if(index < threads.length - 1) {
return threads[index+1];
} else {
return threads[0];
}
}
public void start() {
for (Thread thread : threads) {
thread.start();
}
LockSupport.unpark(threads[0]);
}
}
