并发编程: 管程

张天宇 on 2020-06-07

Java并发编程第二篇,共享模型之管程,各种锁。

1. 共享带来的问题

临界区 Critical Section
  • 一个程序运行多个线程本身是没有问题的
  • 问题出在多个线程访问共享资源
    • 多个线程访问共享资源其实也没有问题
    • 在多个线程对共享资源读写操作时发生指令交错,就会出现问题
  • 一段代码块内如果存在对共享资源的多线程读写操作,称这段代码为临界区
1
2
3
4
5
6
7
8
9
10
static int counter = 0;
static void increment(){
// 临界区
counter++;
}
static void decrement(){
// 临界区
counter--;
}
// 参考JVM内存模型中原子操作类
竞态条件 Race Condition

多个线程在临界区执行,由于代码的执行序列不同而导致结果无法预测,称之为发生了竞态条件。

为了避免临界区的竞态条件发生,有多种手段可以达到目的。

  • 阻塞式的解决方案,synchronized,Lock
  • 非阻塞式的解决方案:原子变量

2. synchronized解决方案

俗称对象锁,它采用互斥的方式让同一时刻至多只有一个线程能持有对象锁,其他线程再想获取这个对象锁时就会被阻塞住,这样就能保证拥有锁的线程可以安全的执行临界区的代码,不用担心上下文的切换。

虽然Java中互斥和同步都可以采用synchronized关键字来完成,但他们还是有区别的:

  • 互斥是保证临界区的竞态条件发生,同一时刻只能有一个线程执行临界区代码
  • 同步是由于线程执行的先后、顺序不同,需要一个线程等待其他线程运行到某个点

Synchonized是基于进入和退出Monitor对象来实现方法同步和代码块同步,但两者的实现细节不一样。Synchronized 用在方法上时,在字节码中是通过方法的 ACC_SYNCHRONIZED 标志来实现的。而代码块同步则是使用monitorenter和monitorexit指令实现的。

monitorenter指令是在编译后插入到同步代码块的开始位置,而monitorexit是插入到方法结束处和异常处,JVM要保证每个monitorenter必须有对应的monitorexit与之配对。任何对象都有一个monitor与之关联,当且一个monitor被持有后,它将处于锁定状态。线程执行到monitorenter指令时,将会尝试获取对象所对应的monitor的所有权,即尝试获得对象的锁,当获得对象的monitor以后,monitor内部的计数器就会自增(初始为0),当同一个线程再次获得monitor的时候,计数器会再次自增。当同一个线程执行monitorexit指令的时候,计数器会进行自减,当计数器为0的时候,monitor就会被释放,其他线程便可以获得monitor。

语法
1
2
3
synchronized (对象){
// 临界区
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
static int counter = 0;
static final Object room = new Object();
public static void main(String[] args) throws InterruptedException{
Thread t1 = new Thread(() -> {
for (int i = 0; i <5000; i++){
synchronized (room){
counter++;
}
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i <5000; i++){
synchronized (room){
counter--;
}
}
});
t1.start();
t2.start();
t1.join();
t2/join();
}

synchronized实际上是用对象锁保证了临界区内代码的原子性,临界区内的代码对外是不可分割的,不会被线程切断所打断。

面向对象改进
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Room room = new Room();
Thread thread = new Thread(() -> {
for (int i = 0; i < 5000; i++)
room.increment();
});
Thread thread1 = new Thread(() -> {
for (int i = 0; i < 5000; i++)
room.decrement();
});
thread.start();
thread1.start();
thread.join();
thread1.join();
log.debug("主线程结束{}", room.getCounter());
}
}
class Room {
private int counter = 0;
public void increment() {
synchronized (this) {
counter++;
}
}
public void decrement() {
synchronized (this) {
counter--;
}
}
public int getCounter() {
synchronized (this) {
return counter;
}
}
}

3. 方法上的synchronized

1
2
3
4
5
6
7
8
9
10
11
class test {
public synchronized void test() {
}
}
// 等价于
class test {
public void test() {
synchronized(this) {
}
}
}
1
2
3
4
5
6
7
8
9
10
11
class test {
public synchronized static void test() {
}
}
// 等价于
class test {
public static void test() {
synchronized(test.class) {
}
}
}
synchronized 关键字修饰方法和代码的区别
  • synchronized修饰方法:同步是隐式的,给方法添加ACC_SYNCHRONIZED的访问修饰符。当某个线程要访问某个方法的时候,会检查是否有ACC_SYNCHRONIZED,如有,则需要先获取当前对象的监视器锁,然后才开始执行方法。方法执行之后再释放监视器锁。
  • 同步代码块:采用monitorentermonitorexit两个指令来实现同步。在执行monitorenter指令时,首先要尝试获取对象的锁。如果这个对象没被锁定,或者当前线程已经拥有了那个对象的锁,把锁的计数器加1,相应的,在执行monitorexit指令时会将锁计数器减1,当计数器为0的时候,锁就会被释放。如果获取对象锁失败,那当前线程就要阻塞等待。直到对象锁被另外一个线程释放为止。

练习:线程八锁

情况1

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n1.b();
}).start();
}
}
@Slf4j
class Number {
public synchronized void a() {
log.debug("1");
}

public synchronized void b() {
log.debug("2");
}
}
/**
两个锁锁的是同一个对象,12或21都可能
2020-06-03 19:28:16.057 DEBUG Test - begin
2020-06-03 19:28:16.059 DEBUG Test - begin
2020-06-03 19:28:16.061 DEBUG Number - 1
2020-06-03 19:28:16.061 DEBUG Number - 2
**/

情况2,修改Number类如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Slf4j
class Number {
public synchronized void a() {
Thred.sleep(1000);
log.debug("1");
}

public synchronized void b() {
log.debug("2");
}
}
/**
此时可能会出现,等一秒,1,释放锁,2
或者2,释放锁,等一秒,1
**/

情况3

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n1.b();
}).start();
new Thread(() -> {
log.debug("begin");
n1.c();
}).start();
}
}
@Slf4j
class Number {
public synchronized void a() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
public void c() {
log.debug("3");
}
}
/** 3, 1s, 12
23/32, 1s, 1
2020-06-03 19:36:22.053 DEBUG Test - begin
2020-06-03 19:36:22.066 DEBUG Test - begin
2020-06-03 19:36:22.067 DEBUG Test - begin
2020-06-03 19:36:22.068 DEBUG Number - 3
2020-06-03 19:36:23.060 DEBUG Number - 1
2020-06-03 19:36:23.060 DEBUG Number - 2
**/

情况4

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
Number n2 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n2.b();
}).start();
}
}
@Slf4j
class Number {
public synchronized void a() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
/**
无互斥发生,2, 1s, 1
**/

情况5

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n1.b();
}).start();
}
}
@Slf4j
class Number {
public static synchronized void a() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
/**
a 锁住的是类对象,b 锁住的是this,不存在互斥的问题,2,1s,1
**/

情况6

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
Number n2 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n1.b();
}).start();
}
}
@Slf4j
class Number {
public static synchronized void a() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("1");
}
public static synchronized void b() {
log.debug("2");
}
}
/**
有互斥情况发生,看CPU先调度谁
**/

情况7

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
Number n2 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n2.b();
}).start();
}
}
@Slf4j
class Number {
public static synchronized void a() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("1");
}
public synchronized void b() {
log.debug("2");
}
}
/**
不存在互斥,并行执行,先2,1s,1
**/

情况8

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Slf4j
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Number n1 = new Number();
Number n2 = new Number();
new Thread(() -> {
log.debug("begin");
n1.a();
}).start();
new Thread(() -> {
log.debug("begin");
n2.b();
}).start();
}
}
@Slf4j
class Number {
public static synchronized void a() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("1");
}
public static synchronized void b() {
log.debug("2");
}
}
/**
有互斥现象存在
**/

4. 线程安全分析

成员变量和静态变量是否线程安全
  • 如果它们没有共享,则线程安全
  • 如果它们被共享了,根据它们的状态是否能被改变,又分两种情况:
    • 如果只有读操作,则线程安全
    • 如果只有写操作,则这段代码属于临界区,需要考虑线程安全
局部变量是否线程安全
  • 局部变量是线程安全的
  • 但局部变量引用的对象未必
    • 如果该对象没有逃离方法的作用域访问,它是线程安全的
    • 如果该对象逃离方法的作用范围,需要考虑线程安全
局部变量线程安全分析
1
2
3
4
public static void text(){
int i = 10;
i++;
}

每个线程调用test()方法时局部变量i,会在每个线程的栈帧内存中被创建多份,因此不存在共享。

但局部变量的引用稍有不同。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@Slf4j
public class Test {
static final int THREAD_NUMBER = 2;
static final int LOOP_NUMBER = 200;
public static void main(String[] args) {
ThreadUnsafe test = new ThreadUnsafe();
for (int i = 0; i < THREAD_NUMBER; i++) {
new Thread(() -> {
test.method1(LOOP_NUMBER);
}, "Thread" + i).start();
}
}
}
class ThreadUnsafe {
ArrayList<String> list = new ArrayList<>();
public void method1(int loopNumber) {
for (int i = 0; i < loopNumber; i++) {
// 临界区, 会产生竞态条件
method2();
method3();
// 临界区
}
}
private void method2() {
list.add("1");
}
private void method3() {
list.remove(0);
}
}
/**
其中一种情况是,如果线程2还未add,线程1remove就会报错
Exception in thread "Thread0" java.lang.IndexOutOfBoundsException: Index: 0, Size: 1
**/

无论哪个线程中的method2引用的都是同一个对象中的list成员变量,method3与method2分析相同。

修改为局部变量,

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@Slf4j
public class Test {
static final int THREAD_NUMBER = 2;
static final int LOOP_NUMBER = 200;

public static void main(String[] args) {
ThreadSafe test = new ThreadSafe();
for (int i = 0; i < THREAD_NUMBER; i++) {
new Thread(() -> {
test.method1(LOOP_NUMBER);
}, "Thread" + i).start();
}
}
}
class ThreadSafe {
public final void method1(int loopNumber) {
ArrayList<String> list = new ArrayList<>();
for (int i = 0; i < loopNumber; i++) {
// { 临界区, 会产生竞态条件
method2(list);
method3(list);
// } 临界区
}
}
private void method2(ArrayList<String> list) {
list.add("1");
}
private void method3(ArrayList<String> list) {
list.remove(0);
}
}
  • list是局部变量,每个线程调用时会创建其不同实例,没有共享
  • 而method2的参数是由method1中传递过来的,与method1引用同一个对象
  • method3的参数分析与method2相同
常见线程安全类
  • String
  • Integer
  • StringBuffer
  • Random
  • Vector
  • Hashtable
  • java.util.concurrent下的类

这里说它们是线程安全的是指,多个线程调用它们同一个实例的某个方法,是线程安全的,也可以理解为:

  • 它们的每个方法是原子的
  • 但是它们多个方法组合不是原子的
不可变线程安全性

String、Integer等都是不可变类,因为其内部的状态不可改变,因此它们的方法都是线程安全的。

实例分析
例1
1
2
3
4
5
6
7
8
9
10
11
12
public class MyServlet extends HttpServlet {
// 多个请求读写内容造成混乱,不是线程安全问题
Map<String, Object> map = new HashMap<>();
// 不可变类,线程安全
String S1 = "...";
// 线程安全
final String S2 = "...";
// 不是线程安全
Date Da = new Date();
// 仍然不是
final Date D2 = new Date();
}
例2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
ublic class MyServlet extends HttpServlet {
// 非线程安全
private UserService userService = new UserServiceImpl();
public void doGet(HttpServletRequest request, HttpServletResponse response) {
userService.update(...);
}
}
public class UserServiceImpl implements UserService {
// 记录调用次数
private int count = 0;
public void update() {
// ...
// 临界区
count++;
}
}
例3
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Aspect
@Component
public class MyAspect {
// 不加Scope说明是单例,会被共享,会存在对象的并发修改,使用环绕通知
private long start = 0L;
@Before("execution(* *(..))")
public void before() {
start = System.nanoTime();
}
@After("execution(* *(..))")
public void after() {
long end = System.nanoTime();
System.out.println("cost time:" + (end-start));
}
}
例4
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public class MyServlet extends HttpServlet {
// 线程安全,因为private私有,没有其他地方修改
private UserService userService = new UserServiceImpl();
public void doGet(HttpServletRequest request, HttpServletResponse response) {
userService.update(...);
}
}
public class UserServiceImpl implements UserService {
// 虽然会被共享,但是userDao里面没有可以被更改的属性了,所以是线程安全的
private UserDao userDao = new UserDaoImpl();
public void update() {
userDao.update();
}
}
public class UserDaoImpl implements UserDao {
// 没有成员变量的类一般是线程安全的
public void update() {
String sql = "update user set password = ? where username = ?";
// 方法内局部变量,线程安全
try (Connection conn = DriverManager.getConnection("","","")){
// ...
} catch (Exception e) {
// ...
}
}
}
例5
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class MyServlet extends HttpServlet {
private UserService userService = new UserServiceImpl();
public void doGet(HttpServletRequest request, HttpServletResponse response) {
userService.update(...);
}
}
public class UserServiceImpl implements UserService {
private UserDao userDao = new UserDaoImpl();
public void update() {
userDao.update();
}
}
public class UserDaoImpl implements UserDao {
// 不是方法内的局部变量,而是方法变量,会被多个线程所共享
private Connection conn = null;
public void update() throws SQLException {
String sql = "update user set password = ? where username = ?";
conn = DriverManager.getConnection("","","");
// ...
conn.close();
}
}
例6
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class MyServlet extends HttpServlet {
private UserService userService = new UserServiceImpl();
public void doGet(HttpServletRequest request, HttpServletResponse response) {
userService.update(...);
}
}
public class UserServiceImpl implements UserService {
public void update() {
// 作为方法内局部变量出现,connection不会出现线程安全问题
UserDao userDao = new UserDaoImpl();
userDao.update();
}
}
public class UserDaoImpl implements UserDao {
private Connection = null;
public void update() throws SQLException {
String sql = "update user set password = ? where username = ?";
conn = DriverManager.getConnection("","","");
// ...
conn.close();
}
}
例7
1
2
3
4
5
6
7
8
9
10
11
public abstract class Test {
public void bar() {
// 是否安全
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
foo(sdf);
}
public abstract foo(SimpleDateFormat sdf);
public static void main(String[] args) {
new Test().bar();
}
}

其中foo的行为是不确定的,可能导致不安全的发生,被称之为外星方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
public void foo(SimpleDateFormat sdf) {
String dateStr = "1999-10-11 00:00:00";
for (int i = 0; i < 20; i++) {
// 在一个新的线程中使用
new Thread(() -> {
try {
sdf.parse(dateStr);
} catch (ParseException e) {
e.printStackTrace();
}
}).start();
}
}

6. Monitor概念

被翻译为监视器或者管程

每个Java对象都可以关联一个Monitor对象,如果使用synchronized给对象上锁(重量级)之后,该对象头的Mark Word中就被设置指向Monitor对象的指针。

  • 刚开始Monitor中Owner为null
  • 当Thread-2执行synchronized(obj)就会将Monitor的所有者Owner置为Thread-2,Monitor中只能有一个Owner
  • 在Thread-2上锁的过程中,如果Thread-3、Thread-4、Thread-5也来执行synchronized(obj),就会进入EntryList BLOCKED
  • Thread-2执行完同步代码块的内容,然后唤醒EntryList中等待的线程来竞争锁,竞争的时候是非公平的

注意

  • synchronized必须是进入同一个对象的monitor才有上述的效果
  • 不加synchronized的对象不会监视器,不遵从以上规则
synchronized进阶
1
2
3
4
5
6
7
8
9
10
11
12
static final Object obj = new Object();
public static void method1() {
synchronized( obj ) {
// 同步块 A
method2();
}
}
public static void method2() {
synchronized( obj ) {
// 同步块 B
}
}
  • 创建锁记录(LockRecord)对象,每个线程的栈帧都会包括一个锁记录的结构,内部可以存储锁定对象的MarkWord

  • 让锁记录中Object Reference指向锁对象,并尝试用CAS(Compare and swap)替换Object的MarkWord,将MarkWord的值存入锁记录

  • 如果CAS替换成功,对象头存储了锁记录地址和状态00,表示由线程给该对象加锁,这时图示如下

  • 如果CAS失败,有两种情况

    • 如果是其他线程已经持有了该Object的轻量级锁,这时候表明有竞争,进入锁膨胀过程

    • 如果是自己执行了synchronized锁重入,那么再添一条LockRecord作为重入的计数

  • 当退出synchronized代码块(解锁时)如果有取值为null的锁记录,表示有重入,这时重置锁记录,表示重入计数减一

  • 当退出synchronized代码块(解锁时)锁记录的值不为null,这时使用cas将MarkWord的值恢复给对象头

    • 成功,则解锁成功
    • 失败,说明轻量级锁进行了锁膨胀或已经升级为重量锁,进入重量级锁解锁流程
锁膨胀

如果在尝试加轻量锁的过程中,CAS操作无法成功,这时一种情况就是有其他线程为此对象加上了轻量级锁(有竞争),这时需要进行锁膨胀,将轻量级锁变为重量级锁。

1
2
3
4
5
6
static Object obj = new Object();
public static void method1() {
synchronized( obj ) {
// 同步块
}
}
  • 当Thread-1进行轻量级加锁时,Thread-0已经对该对象加了轻量级锁

  • 这时Thread-1加轻量级锁失败,进入锁膨胀流程

    • 即为Object对象申请Monitor锁,让Object指向重量级锁地址

    • 然后自己进入Monitor的EntryList BLOCKED

  • 当Thread-0退出同步代码块解锁时,使用CAS将MarkWord的值恢复给对象头,失败,这时会进入重量级解锁流程,即按照Monitor地址找到Monitor对象,设置Owner为null,唤醒EntryList中BLOCKED线程

自旋优化

重量级锁竞争的时候,还可以使用自旋来优化,如果当前线程自旋成功(即这时候锁线程已经退出了同步块,释放了锁),这时当前线程就可以避免阻塞。

  • 自旋重试成功的情况
  • 失败

偏向锁

轻量级锁在没有竞争时(就自己这个线程),每次重入仍然需要执行CAS操作。

Java6中引入了偏向锁来做进一步优化,只有第一次使用CAS将线程ID设置到对象的MarkWord头,之后发现这个线程ID是自己的就表示没有竞争,不用重新CAS,以后只要不发生竞争,这个对象就归该线程所有。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
static final Object obj = new Object();
public static void m1() {
synchronized( obj ) {
// 同步块 A
m2();
}
}
public static void m2() {
synchronized( obj ) {
// 同步块 B
m3();
}
}
public static void m3() {
synchronized( obj ) {
// 同步块 C
}
}
偏向状态

一个对象创建时:

  • 如果开启了偏向锁(默认开启),那么对象创建后,markword值为0x05即后三位为101,这时他的thread、epoch、age都为0
  • 偏向锁默认是延迟的,不会在程序启动时立即生效,如果想避免延迟,可以加JVM参数-XX:biasedLockingStartupDelay=0来禁用延迟
  • 如果没有开启偏向锁,那么对象创建后,markword值为0x01即后三位为001,这时它的hashcode、age都为0,第一次用到hashcode时候才会赋值
  • -XX:-UseBiasedLocking,禁用偏向锁
  • 当一个可偏向的对象,调用了自己的hashcode之后,会撤销他的偏向状态
撤销-调用hashcode

调用了对象的hashcode,但偏向锁的对象markword中存储的是线程id,如果调用hashCode会导致偏向锁被撤销。

  • 轻量级锁会在锁记录中记录hashCode
  • 重量级锁会在Monitor中记录hashCode

在调用hashCode后使用偏向锁,记得去掉-XX:-UseBiasedLocking

撤销-其他线程使用对象

当有其他线程使用偏向锁对象时,会将偏向锁升级为轻量级锁,由可偏向变为不可偏向。

撤销-调用wait/notify

因为wait/notify只有重量级锁才有,因此调用了就会升级为重量级锁。

批量重偏向

如果对象虽然被多个线程访问,但没有竞争,这时偏向了线程T1的对象仍然有机会重新偏向T2,重偏向会重置对象的ThreadID。

当撤销偏向锁阈值超过二十次以后,JVM会觉得自己是不是偏向错了,于是会给这些对象加锁时重新偏向至加锁线程。

批量撤销

当撤销偏向锁阈值超过40次后,JVM会觉得自己确实偏向错了,根本就不该偏向,于是整个类的所有对象都会变为不可偏向的,新建的对象也是不可偏向的。

锁消除
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Fork(1)
@BenchmarkMode(Mode.AverageTime)
@Warmup(iterations=3)
@Measurement(iterations=5)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
public class MyBenchmark {
static int x = 0;
@Benchmark
public void a() throws Exception {
x++;
}
@Benchmark
public void b() throws Exception {
Object o = new Object();
synchronized (o) {
x++;
}
}
}
1
2
3
4
# java -jar benchmarks.jar 热点代码优化,性能相差无几
Benchmark Mode Samples Score Score error Units
c.i.MyBenchmark.a avgt 5 1.542 0.056 ns/op
c.i.MyBenchmark.b avgt 5 1.518 0.091 ns/op
1
2
3
4
# java -XX:-EliminateLocks -jar benchmarks.jar 关闭锁消除优化
Benchmark Mode Samples Score Score error Units
c.i.MyBenchmark.a avgt 5 1.507 0.108 ns/op
c.i.MyBenchmark.b avgt 5 16.976 1.572 ns/op

锁粗化:对相同对象多次加锁,导致线程发生多次重入,可以使用锁粗化方式来优化,这不同于之前讲的细分锁的粒度。

7. wait / notify

  • Owner线程发现条件不满足,调用wait方法,即可进入WaitSet变为Waiting状态
  • BLOCKED和WAITING的线程都处于阻塞状态,不占用CPU时间片
  • BLOCKED线程会在Owner线程释放锁时唤醒
  • WAITING线程会在Owner线程调用notify或notufyAll时候唤醒, 但唤醒后并不意味着立刻获得锁,仍需进入EntryList重新竞争
API
  • obj.wait()让进入object监视器的线程到waitSet等待
  • obj.notify()在object上正在waitSet等待的线程中挑一个唤醒
  • obj.notifyAll()让object上正在waitSet等待的线程全部唤醒

它们都是线程之间进行协作的手段,都属于Object对象的方法,必须获得此对象的锁,才能调用这几个方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
final static Object obj = new Object();
public static void main(String[] args) {
new Thread(() -> {
synchronized (obj) {
log.debug("执行....");
try {
obj.wait(); // 让线程在obj上一直等待下去
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("其它代码....");
}
}).start();
new Thread(() -> {
synchronized (obj) {
log.debug("执行....");
try {
obj.wait(); // 让线程在obj上一直等待下去
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("其它代码....");
}
}).start();
// 主线程两秒后执行
sleep(2);
log.debug("唤醒 obj 上其它线程");
synchronized (obj) {
obj.notify(); // 唤醒obj上一个线程
// obj.notifyAll(); // 唤醒obj上所有等待线程
}
}
1
2
3
4
5
# notify的一种结果
20:00:53.096 [Thread-0] c.TestWaitNotify - 执行....
20:00:53.099 [Thread-1] c.TestWaitNotify - 执行....
20:00:55.096 [main] c.TestWaitNotify - 唤醒 obj 上其它线程
20:00:55.096 [Thread-0] c.TestWaitNotify - 其它代码....
1
2
3
4
5
6
# notifyAll的结果
19:58:15.457 [Thread-0] c.TestWaitNotify - 执行....
19:58:15.460 [Thread-1] c.TestWaitNotify - 执行....
19:58:17.456 [main] c.TestWaitNotify - 唤醒 obj 上其它线程
19:58:17.456 [Thread-1] c.TestWaitNotify - 其它代码....
19:58:17.456 [Thread-0] c.TestWaitNotify - 其它代码....

wait()方法会释放对象的锁,进入WaitSet等待区,从而让其他线程有机会获得对象的锁。无限制等待直到notify为止。

wait(long n)有时限的等待,到n毫秒后结束等待,或被notify,wait(long timeout, int nanos)纳秒。

8. wait/notify的正确姿势

sleep和notify的区别
  • sleep是Thread方法,而wait是Object的方法
  • sleep不需要强制和synchronized配合使用,但wait需要和synchronized使用
  • sleep在睡眠的同时,不会释放对象锁的,但wait在等待的时候会释放对象锁
  • 它们的状态都是TIMED_WAITING
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
new Thread(() -> {
synchronized (room) {
log.debug("有烟没?[{}]", hasCigarette);
while (!hasCigarette) { // 使用while解决虚假唤醒问题
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();

套路总结

1
2
3
4
5
6
7
8
9
10
synchronized(lock) {
while(条件不成立) {
lock.wait();
}
// 干活
}
//另一个线程
synchronized(lock) {
lock.notifyAll();
}
同步模式之保护性暂停

Guarded Suspension,用在一个线程等待另一个线程的执行结果。

  • 有一个结果需要从一个线程传递到另一个线程,让它们关联同一个GuardedObject
  • 如果有结果不断从一个线程到另一个线程,可以使用消息队列
  • JDK中,join的实现、Future的实现,采用的就是此模式
  • 因为要等待另一方的结果,因此归类到同步模式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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();
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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());
}
增加超时效果
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class GuardedObjectV {
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();
}
}
}
join原理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public final void join() throws InterruptedException {
join(0);
}
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()) {
wait(0);
}
} else {
while (isAlive()) {
long delay = millis - now;
if (delay <= 0) {
break;
}
wait(delay);
now = System.currentTimeMillis() - base;
}
}
}
解耦

以居民邮箱为例,在多个类之间进行参数传递,设计一个用来解耦的中间类,解耦结果等待者和结果生产者。

生产线程和结果使用线程是一一对应的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
@Slf4j
public class Test {
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 3; i++) {
new People().start();
}
Thread.sleep(1000);
for (Integer id : MailBoxes.getIds()) {
new Postman(id, "内容" + id).start();
}
}
}
//业务相关类
@Slf4j
class People extends Thread {
@Override
public void run() {
// 收信操作
GuardedObject guardedObject = MailBoxes.createGuardedObject();
log.debug("开始收信 id:{}", guardedObject.getId());
Object mail = guardedObject.get(5000);
log.debug("收到信 id:{} 内容:{}", guardedObject.getId(), mail);
}
}
@Slf4j
class Postman extends Thread {
private int id;
private String mail;

public Postman(int id, String mall) {
this.id = id;
this.mail = mall;
}
@Override
public void run() {
GuardedObject guardedObject = MailBoxes.getGuardedObject(id);
log.debug("送信 id:{}, 内容:{}", id, mail);
guardedObject.complete(mail);
}
}
// 用于解耦的通用类
class MailBoxes {
public static Map<Integer, GuardedObject> boxes = new Hashtable<>();
private static int id = 1;
private static synchronized int generateId() {
return id++;
}
public static GuardedObject getGuardedObject(int id) {
return boxes.remove(id);
}
public static GuardedObject createGuardedObject() {
GuardedObject go = new GuardedObject(generateId());
boxes.put(go.getId(), go);
return go;
}
public static Set<Integer> getIds() {
return boxes.keySet();
}
}
class GuardedObject {
// 新建一个id用来表示GuardedObject
private int id;
private Object response;
public GuardedObject(int id) {
this.id = id;
}
public int getId() {
return id;
}
public Object get(long timeout) {
synchronized (this) {
long begin = System.currentTimeMillis();
long passedTime = 0;
while (response == null) {
long waitTime = timeout - passedTime;
if (timeout - passedTime <= 0) {
break;
}
try {
this.wait(waitTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
passedTime = System.currentTimeMillis() - begin;
}
return response;
}
}
public void complete(Object response) {
synchronized (this) {
this.response = response;
this.notifyAll();
}
}
}
/**
2020-06-07 10:59:22.336 DEBUG People - 开始收信 id:3
2020-06-07 10:59:22.336 DEBUG People - 开始收信 id:1
2020-06-07 10:59:22.336 DEBUG People - 开始收信 id:2
2020-06-07 10:59:23.368 DEBUG Postman - 送信 id:3, 内容:内容3
2020-06-07 10:59:23.369 DEBUG People - 收到信 id:3 内容:内容3
2020-06-07 10:59:23.369 DEBUG Postman - 送信 id:2, 内容:内容2
2020-06-07 10:59:23.369 DEBUG People - 收到信 id:2 内容:内容2
2020-06-07 10:59:23.370 DEBUG Postman - 送信 id:1, 内容:内容1
2020-06-07 10:59:23.370 DEBUG People - 收到信 id:1 内容:内容1
**/
异步模式之生产者消费者
  • 与保护性暂停中的GuardedObject不同,不需要产生结果和消费结果的线程一一对应
  • 消费队列可以用来平衡生产和消费的线程资源
  • 生产者仅负责产生结果数据,不关心数据该如何处理,而消费者专心处理结果数据
  • 消息队列是有容量限制的,满时不会再加入数据,空时不会再消耗数据
  • JDK中各种阻塞队列,采用的就是这种模式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@Slf4j
public class Test {
public static void main(String[] args) throws InterruptedException {
MessageQueue messageQueue = new MessageQueue(2);
for (int i = 0; i < 3; i++) {
int id = i;
new Thread(() -> {
messageQueue.put(new Message(id, "值" + id));
}, "生产者" + i).start();
}
new Thread(() -> {
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
Message message = messageQueue.take();
}
}, "消费者").start();
}
}
@Slf4j
class MessageQueue {
private LinkedList<Message> queue;
private int capacity;

MessageQueue(int capacity) {
this.capacity = capacity;
queue = new LinkedList<>();
}
public Message take() {
synchronized (queue) {
while (queue.isEmpty()) {
log.debug("没货了,wait");
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
Message message = queue.removeFirst();
log.debug("已经消费了一个消息");
queue.notifyAll();
return message;
}
}
public void put(Message message) {
synchronized (queue) {
while (queue.size() == capacity) {
log.debug("库存已经达到上限,wait");
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.addLast(message);
log.debug("已经生产了一个消息");
queue.notifyAll();
}
}
}
final class Message {
private int id;
private Object message;
Message(int id, Object message) {
this.id = id;
this.message = message;
}
public int getId() {
return id;
}
public Object getMessage() {
return message;
}
}
/**
2020-06-07 14:58:19.298 DEBUG MessageQueue - 已经生产了一个消息
2020-06-07 14:58:19.303 DEBUG MessageQueue - 已经生产了一个消息
2020-06-07 14:58:19.303 DEBUG MessageQueue - 库存已经达到上限,wait
2020-06-07 14:58:20.296 DEBUG MessageQueue - 已经消费了一个消息
2020-06-07 14:58:20.297 DEBUG MessageQueue - 已经生产了一个消息
2020-06-07 14:58:21.297 DEBUG MessageQueue - 已经消费了一个消息
2020-06-07 14:58:22.297 DEBUG MessageQueue - 已经消费了一个消息
2020-06-07 14:58:23.297 DEBUG MessageQueue - 没货了,wait
**/

9. Park & Unpark

基本使用
1
2
3
4
5
// 它们是LockSupport类的方法
// 暂停当前线程
LockSupport.park();
// 恢复某个线程的运行
LockSupport.unpark(暂停线程的对象);
1
2
3
4
5
6
7
8
9
10
11
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);
特点
  • wait、notify和notifyAll必须配合Object Monitor一起使用,而park、unpark不必
  • park & unpark以线程为单位来阻塞和唤醒线程,而notify只能随机唤醒一个等待线程,notifyAll是唤醒所有等待线程,不是那么精确
  • park & unpark可以先unpark,而wait & notify不能先notify
原理

每个线程都有自己的一个Parker对象,由三部分组成,_counter_cond和*_mutex*,打个比喻。

  • 线程就像一个旅人,Parker就像他随身携带的背包,条件变量就好比背包中的帐篷,_counter就好比背包中的备用干粮(0为耗尽,1为充足)
  • 调用park就是要看需不需要停下来歇息
    • 如果备用干粮耗尽,那么钻进帐篷休息
    • 如果备用干粮充足,那么不需停留,继续前进
  • 调用unpark,就好比令干粮充足
    • 如果这时线程还在帐篷,就唤醒让他继续前进
    • 如果这时线程还在运行,那么他下次调用park时,仅是消耗备用干粮,不需停留继续前进
      • 因为背包空间有限,多次调用unpark仅会补充一份备用干粮

10. 重新理解线程状态转换




11. 多把锁

一间屋子有学习和睡觉两个功能,如果A要学习,B要睡觉,但是如果只用一个屋子即一个对象锁的话,并发度很低,解决办法就是转杯多个房间即多个对象锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class BigRoom {
public void sleep() {
synchronized (this) {
log.debug("sleeping 2 小时");
Sleeper.sleep(2);
}
}
public void study() {
synchronized (this) {
log.debug("study 1 小时");
Sleeper.sleep(1);
}
}
}

改进

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class BigRoom {
private final Object studyRoom = new Object();
private final Object bedRoom = new Object();
public void sleep() {
synchronized (bedRoom) {
log.debug("sleeping 2 小时");
Sleeper.sleep(2);
}
}
public void study() {
synchronized (studyRoom) {
log.debug("study 1 小时");
Sleeper.sleep(1);
}
}
}

将锁的细粒度细分

  • 好处:增强并发度
  • 坏处:如果一个线程需要同时获得多把锁,容易发生死锁

12. 活跃性

死锁

一个线程需要同时获得多把锁,这时候就容易发生死锁。

  • T1线程获得A对象锁,接下来想获得B对象锁
  • T2线程获得B对象锁,接下来想获得A对象锁

使用顺序加锁的方式解决该问题,即都按照AB的顺序获得锁。

定位死锁
  • 检测死锁可以使用jsconsole工具,或者使用jps定位进程id,再用jstact定位死锁
  • 避免死锁要注意加锁顺序
  • 如果某个进程进入了死循环,导致其它线程一直等待,对于这种情况,Linux下可以通过top先定位到CPU占用高的Java进程,再利用top -Hp 进程id来定位哪个进程,最后再用jstack排查
活锁

活锁出现在两个线程互相改变对方的结束条件,最后谁也无法结束。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class TestLiveLock {
static volatile int count = 10;
static final Object lock = new Object();
public static void main(String[] args) {
new Thread(() -> {
// 期望减到 0 退出循环
while (count > 0) {
sleep(0.2);
count--;
log.debug("count: {}", count);
}
}, "t1").start();
new Thread(() -> {
// 期望超过 20 退出循环
while (count < 20) {
sleep(0.2);
count++;
log.debug("count: {}", count);
}
}, "t2").start();
}
}
饥饿

一个线程由于优先级太低,始终得不到CPU调度执行,也不能结束。

13. ReentrantLock

可重入锁

  • 可中断
  • 可以设置超时时间
  • 可以设置公平锁
  • 支持多个条件变量

与synchronized一样,都支持可重入。

1
2
3
4
5
6
7
8
9
// 先创建一个ReentrantLock对象
// 获取锁
reentrantLock.lock();
try{
// 临界区
} finally {
// 释放锁
reentrantLock.unlock();
}
可重入

可重入是指同一个线程如果首次获得了这把锁,那么是因为它是这把锁的拥有者,因此有权利再次获取这把锁。

如果是不可重入锁,那么第二次获得锁时,自己也会被锁挡住。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {
method1();
}
public static void method1() {
lock.lock();
try {
log.debug("execute method1");
method2();
} finally {
lock.unlock();
}
}
public static void method2() {
lock.lock();
try {
log.debug("execute method2");
method3();
} finally {
lock.unlock();
}
}
public static void method3() {
lock.lock();
try {
log.debug("execute method3");
} finally {
lock.unlock();
}
}
/**
17:59:11.862 [main] c.TestReentrant - execute method1
17:59:11.865 [main] c.TestReentrant - execute method2
17:59:11.865 [main] c.TestReentrant - execute method3
**/
可打断
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
ReentrantLock lock = new ReentrantLock();
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 {
sleep(1);
t1.interrupt();
log.debug("执行打断");
} finally {
lock.unlock();
}
/**
18:02:40.520 [main] c.TestInterrupt - 获得了锁
18:02:40.524 [t1] c.TestInterrupt - 启动...
18:02:41.530 [main] c.TestInterrupt - 执行打断
java.lang.InterruptedException
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchr
onizer.java:898)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchron
izer.java:1222)
at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
at cn.itcast.n4.reentrant.TestInterrupt.lambda$main$0(TestInterrupt.java:17)
at java.lang.Thread.run(Thread.java:748)
18:02:41.532 [t1] c.TestInterrupt - 等锁的过程中被打断
**/

注意如果是不可中断模式lock(),那么即使使用了interrupt也不会让等待中断。

锁超时
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
ReentrantLock lock = new ReentrantLock();
Thread t1 = new Thread(() -> {
log.debug("启动...");
try {
if (!lock.tryLock(1, TimeUnit.SECONDS)) {
log.debug("获取等待 1s 后失败,返回");
return;
}
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
log.debug("获得了锁");
} finally {
lock.unlock();
}
}, "t1");
lock.lock();
log.debug("获得了锁");
t1.start();
try {
sleep(2);
} finally {
lock.unlock();
}
// lock.tryLock()表示不等待直接获得锁

解决哲学家就餐问题

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class Chopstick {
String name;
public Chopstick(String name) {
this.name = name;
}
@Override
public String toString() {
return "筷子{" + name + '}';
}
}
class Philosopher extends Thread {
Chopstick left;
Chopstick right;
public Philosopher(String name, Chopstick left, Chopstick right) {
super(name);
this.left = left;
this.right = right;
}
private void eat() {
log.debug("eating...");
Sleeper.sleep(1);
}
@Override
public void run() {
while (true) {
// 获得左手筷子
synchronized (left) {
// 获得右手筷子
synchronized (right) {
// 吃饭
eat();
}
// 放下右手筷子
}
// 放下左手筷子
}
}
}

// 测试
Chopstick c1 = new Chopstick("1");
Chopstick c2 = new Chopstick("2");
Chopstick c3 = new Chopstick("3");
Chopstick c4 = new Chopstick("4");
Chopstick c5 = new Chopstick("5");
new Philosopher("苏格拉底", c1, c2).start();
new Philosopher("柏拉图", c2, c3).start();
new Philosopher("亚里士多德", c3, c4).start();
new Philosopher("赫拉克利特", c4, c5).start();
new Philosopher("阿基米德", c5, c1).start();
// 当每个人都拿到一只筷子后出现死锁
  1. 调换顺序使方向相同,但是会产生饥饿问题

    1
    new Philosopher("阿基米德", c1, c5).start();
  2. 使用可重入锁解决问题

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    // 使Chopstick继承ReentrantLock
    class Chopstick extends ReentrantLock {}
    // 修改哲学家吃饭逻辑,左手拿到筷子后如果右手拿不到筷子就释放左手中的筷子
    @Override
    public void run() {
    while (true) {
    // 尝试获得左手筷子
    if (left.tryLock()) {
    try {
    // 尝试获得右手筷子
    if (right.tryLock()) {
    try {
    eat();
    } finally {
    right.unlock();
    }
    }
    } finally {
    left.unlock();
    }
    }
    }
    }
公平锁

ReentrantLock默认是不公平的,即构造参数中默认不写为false,设置公平锁ReentrantLock lock = new ReentrantLock(true);

公平锁一般没有必要,会降低并发度。

条件变量

synchronized中也有条件变量,即WaitSet休息室。而ReentrantLock的条件变量要比synchronized强大,它支持多个条件变量。

  • synchronized让不满足条件的线程都在一间休息室等消息
  • ReentrantLock支持多间休息室,有专门等烟的休息室、等早餐的休息室,唤醒时也是按照休息室来唤醒

使用流程

  • await之前需要先获得锁
  • await执行后,会释放锁,进入conditionObject等待
  • await的线程被唤醒(或打断、或超时)去重新竞争lock锁
  • 竞争lock锁成功后,从await后继续执行

以送烟送外卖为例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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) {
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();
sleep(1);
sendBreakfast();
sleep(1);
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();
}
}
同步模式之顺序控制
  1. 控制运行顺序

    • wait/notify版

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      // 用来同步的对象
      static Object obj = new Object();
      // t2 运行标记, 代表 t2 是否执行过
      static boolean t2runed = false;
      public static void main(String[] args) {
      Thread t1 = new Thread(() -> {
      synchronized (obj) {
      // 如果 t2 没有执行过
      while (!t2runed) {
      try {
      // t1 先等一会
      obj.wait();
      } catch (InterruptedException e) {
      e.printStackTrace();
      }
      }
      }
      System.out.println(1);
      });
      Thread t2 = new Thread(() -> {
      System.out.println(2);
      synchronized (obj) {
      // 修改运行标记
      t2runed = true;
      // 通知 obj 上等待的线程(可能有多个,因此需要用 notifyAll)
      obj.notifyAll();
      }
      });
      t1.start();
      t2.start();
      }
      // 可以修改为ReentrantLock重解该问题
    • park/unpark版

      • 首先,需要保证先wait再notify,否则wait线程永远得不到唤醒。因此使用了『运行标记』来判断该不该wait

      • 第二,如果有些干扰线程错误地notify了wait线程,条件不满足时还要重新等待,使用了while循环来解决
        此问题

      • 最后,唤醒对象上的wait线程需要使用notifyAll,因为『同步对象』上的等待线程可能不止一个

        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        Thread t1 = new Thread(() -> {
        try { Thread.sleep(1000); } catch (InterruptedException e) { }
        // 当没有『许可』时,当前线程暂停运行;有『许可』时,用掉这个『许可』,当前线程恢复运行
        LockSupport.park();
        System.out.println("1");
        });
        Thread t2 = new Thread(() -> {
        System.out.println("2");
        // 给线程 t1 发放『许可』(多次连续调用 unpark 只会发放一个『许可』)
        LockSupport.unpark(t1);
        });
        t1.start();
        t2.start();
        // park和unpark方法比较灵活,他俩谁先调用,谁后调用无所谓。并且是以线程为单位进行『暂停』和『恢复』,不需要『同步对象』和『运行标记』
    1. 交替输出

      线程1输出a5次,线程2输出b5次,线程3输出c5次,要求输出abcabcabcabcabc。

      • wait/notify版

        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        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();
        }
        }
        }
        }
        // 测试
        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();
      • ReentrantLock版本

        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43
        44
        45
        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;
        }
        }
        // 测试
        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); // 得先唤醒一个
      • park/unpark版

        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43
        44
        45
        46
        47
        48
        49
        50
        51
        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]);
        }
        }
        // 测试
        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();

场景

交替打印ABABAABA
可重入锁配合 Condition
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ABA {
private static int flag = 0;
private static ReentrantLock lock = new ReentrantLock();
private static Condition aCondition = lock.newCondition();
private static Condition bCondition = lock.newCondition();

static class PrintA implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
lock.lock();
try {
while (flag % 2 != 0) {
try {
aCondition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
flag++;
System.out.print("A");
bCondition.signal();
} finally {
lock.unlock();
}
}
}
}

static class PrintB implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
lock.lock();
try {
while (flag % 2 == 0) {
try {
bCondition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
flag++;
System.out.print("B");
aCondition.signal();
} finally {
lock.unlock();
}
}
}
}

public static void main(String[] args) {
new Thread(new PrintA()).start();
new Thread(new PrintB()).start();
}
}
可重入锁
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
public class ABA {
private static int flag = 0;
private static Lock lock = new ReentrantLock();

static class PrintA implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; ) {
lock.lock();
try {
while (flag % 2 == 0) {
System.out.print("A");
flag++;
i++;
}
} finally {
lock.unlock();
}
}
}
}

static class PrintB implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; ) {
lock.lock();
try {
while (flag % 2 != 0) {
System.out.print("B");
flag++;
i++;
}
} finally {
lock.unlock();
}
}
}
}

public static void main(String[] args) {
new Thread(new PrintA()).start();
new Thread(new PrintB()).start();
}
}
wait() 和 notify()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
public class ABA {
private static int flag = 0;
private Object lock;

private synchronized void printA() {
while (flag % 2 != 0) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
flag++;
System.out.print("A");
notify();
}

private synchronized void printB() {
while (flag % 2 == 0) {
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
flag++;
System.out.print("B");
notify();
}

class PrintA implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
printA();
}
}
}

class PrintB implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
printB();
}
}
}

public static void main(String[] args) {
ABA aba = new ABA();
PrintA printA = aba.new PrintA();
PrintB printB = aba.new PrintB();
new Thread(printA).start();
new Thread(printB).start();
}
}
信号量
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import java.util.concurrent.Semaphore;
public class ABA {
private static Semaphore pa = new Semaphore(1);
private static Semaphore pb = new Semaphore(0);
static class PrintA implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
pa.acquire();
System.out.print("A");
pb.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
static class PrintB implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
pb.acquire();
System.out.print("B");
pa.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) {
new Thread(new PrintA()).start();
new Thread(new PrintB()).start();
}
}
i++ 线程安全性

i++ 是不安全的,它是一个复合操作,可分为三个阶段:

  • 读值,从内存到寄存器
  • +1,寄存器自增
  • 写值,写回内存

这三步之间都可能会有 CPU 调度,造成值被修改,造成脏读和脏写。如果是方法内定义的,一定是线程安全的,因为每个方法栈是线程私有的;如果是类的静态成员变量,则不是线程安全的,因为线程共享栈区,不共享堆区和全局区。

用 Volatile 虽然能保证可见性,但是不能保证原子性,如果想要保证多线程下的安全性,可以使用原子变量(AtomicInteger)、synchronized 和 Lock锁实现。AtomicInteger 和各种 Lock 都可以确保线程安全,AtomicInteger 的效率高是因为它的互斥区非常小,而 Lock 的互斥区是拿到锁到放锁之间的区域,至少三条指令。

Java 线程同步的几种方法

  1. synchronized
  2. wait() 和 notify()
  3. volatile
  4. reentrantlock
  5. ThreadLocal 局部变量
  6. 使用阻塞队列BlockingQueue
  7. 使用原子变量 AtomicInteger