Java多线程-ReentrantLock

相对于synchronized,它具备:

  • 可中断
  • 可设置超时时间
  • 可设置为公平锁
  • 支持多个条件变量(多个休息室)

它和synchronized一样支持可重入

基本语法格式是:

1
2
3
4
5
6
7
8
9
10
private static ReentrantLock lock = new ReentrantLock();
//获取锁
lock.lock();
try{
//临界区
System.out.println("t1获得锁");
}finally {
//释放锁
lock.unlock();
}

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
32
package juc.thread;

import java.util.concurrent.locks.ReentrantLock;

/**
* @Author qq
* @Date 2022/3/18
*/
public class reentrantLock {
private static ReentrantLock lock = new ReentrantLock();

public static void main(String[] args) {
m1();
}
public static void m1(){
lock.lock();
try{
System.out.println("进入了m1");
m2();
}finally {
lock.unlock();
}
}
public static void m2(){
lock.lock();
try{
System.out.println("进入了m2");
}finally {
lock.unlock();
}
}
}

2.可中断的

被动,必须由其他线程执行那个打断方法才能让这个线程不死等下去。

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
package juc.thread;

import java.util.concurrent.locks.ReentrantLock;

/**
* @Author qq
* @Date 2022/3/18
*/
public class reentrantLock {
private static ReentrantLock lock = new ReentrantLock();

public static void main(String[] args) {
Thread thread = new Thread(() -> {
lock.lock();
try{
System.out.println("尝试获取锁");
//获得了锁,此时可以被其他线程打断
lock.lockInterruptibly();
}catch (InterruptedException e){
System.out.println("被打断");
e.printStackTrace();
}
try{
System.out.println("获取到锁");
} finally {
lock.unlock();
}
});
lock.lock();
try{
thread.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}finally {
lock.unlock();
}
thread.interrupt();
}

}

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
package juc.thread;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
* @Author qq
* @Date 2022/3/18
*/
public class reentrantLock {
private static ReentrantLock lock = new ReentrantLock();

public static void main(String[] args) throws InterruptedException {
Thread thread = new Thread(() -> {

System.out.println("尝试获取锁");
//尝试获得锁,如果没有获得,会进入阻塞队列,等待1秒,如果还没有获得到锁就不等了
try {
if(!lock.tryLock(1, TimeUnit.SECONDS)){
System.out.println("等了1s,获取不到锁,退出");
return;
}
} catch (InterruptedException e) {
e.printStackTrace();
}
try{
System.out.println("获取到锁");
}finally {
lock.unlock();
}
});
lock.lock();
Thread.sleep(2000);
thread.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
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
package juc.thread;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
* @Author qq
* @Date 2022/3/16
*/
public class Philosopher extends Thread {

chopsticks left;
chopsticks right;
String name;
public Philosopher(String name,chopsticks left,chopsticks right){
this.name = name;
this.right = right;
this.left = left;
}
@Override
public void run() {
while(true){
if(left.tryLock()){
try{
if(right.tryLock()){
try{
eat();
}finally {
right.unlock();
}
}
}finally {//只有左手筷子,则会放下筷子
left.unlock();
}
}
}


}
public void eat(){
System.out.println(name + "拿上筷子" + left.name + "," + right.name + "开始吃饭..");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "放下筷子" + left.name + "," + right.name + "开始思考..");
}

public static void main(String[] args) {
chopsticks c1 = new chopsticks("c1");
chopsticks c2 = new chopsticks("c2");
chopsticks c3 = new chopsticks("c3");
chopsticks c4 = new chopsticks("c4");
chopsticks c5 = new chopsticks("c5");
Philosopher p1 = new Philosopher("哲学家1",c1,c2);
Philosopher p2 = new Philosopher("哲学家2",c2,c3);
Philosopher p3 = new Philosopher("哲学家3",c3,c4);
Philosopher p4 = new Philosopher("哲学家4",c4,c5);
Philosopher p5 = new Philosopher("哲学家5",c5,c1);
p1.start();
p2.start();
p3.start();
p4.start();
p5.start();
}
}
class chopsticks extends ReentrantLock {
String name;
public chopsticks(String name){
this.name = name;
}
}

4.条件变量

创建新的条件变量:Condition c1 = lock.newCondition(); Condition c2 = lock.newCondition();

阻塞前必须先获得锁:lock.lock();

进入休息室等待:c1.await();

由其他线程唤醒这个休息室中的线程:c1.signal();

5.原理

5.1 构造器

默认实现是非公平锁

1
2
3
public ReentrantLock() {
this.sync = new ReentrantLock.NonfairSync();
}

5.2 加锁

1
2
3
public void lock() {
this.sync.acquire(1);//调用同步器类的方法
}
1
2
3
4
5
public final void acquire(int arg) {
if (!this.tryAcquire(arg) && this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), arg)) {
selfInterrupt();
}
}

this.tryAcquire(arg):尝试去加锁,如果加锁成功就不会进入if语句块,如果加锁失败,尝试创建一个Node对象,将这个线程关联到这个Node对象上,让这个线程进入Node阻塞队列

灰色表示线程正式被阻塞,三角形内的-1表示这个Node对象关联的线程有责任去唤醒它的后继结点,处于链表尾部的节点上面的三角形值是0,它不需要唤醒任何节点

5.3 释放锁

1
2
3
public void unlock() {
this.sync.release(1);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
public final boolean release(int arg) {
if (this.tryRelease(arg)) {
//队列中的头结点,那个虚拟节点
AbstractQueuedSynchronizer.Node h = this.head;
if (h != null && h.waitStatus != 0) {
//唤醒后继节点
this.unparkSuccessor(h);
}

return true;
} else {
return false;
}
}

没有其他新的线程来竞争,那队列的第一个有线程关联的节点的线程就成功获得锁。

但是如果有新的线程来竞争,就不一定谁获得锁了,如果这时队列中第一个线程没有成功获得锁,就需要继续阻塞等待。

5.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
@ReservedStackAccess
final boolean nonfairTryAcquire(int acquires) {
Thread current = Thread.currentThread();
//获取当前锁的状态,为0表示还没有被任何线程持有
int c = this.getState();
if (c == 0) {
//将状态改为1,说明有线程获得了锁
if (this.compareAndSetState(0, acquires)) {
//将锁的持有者改为current线程
this.setExclusiveOwnerThread(current);
return true;
}
//当前线程就是锁的持有者
} else if (current == this.getExclusiveOwnerThread()) {
//state++
int nextc = c + acquires;
if (nextc < 0) {
throw new Error("Maximum lock count exceeded");
}

this.setState(nextc);
return true;
}

return false;
}

重入释放锁:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@ReservedStackAccess
protected final boolean tryRelease(int releases) {
int c = this.getState() - releases;
if (Thread.currentThread() != this.getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException();
} else {
boolean free = false;
//c为0表示线程完全释放了锁
if (c == 0) {
free = true;
this.setExclusiveOwnerThread((Thread)null);
}

this.setState(c);
return free;
}
}

5.5 可打断

lock.lock();默认是不可打断的

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
final boolean acquireQueued(AbstractQueuedSynchronizer.Node node, int arg) {
boolean interrupted = false;

try {
while(true) {
AbstractQueuedSynchronizer.Node p = node.predecessor();
if (p == this.head && this.tryAcquire(arg)) {
this.setHead(node);
p.next = null;
return interrupted;
}
//如果有线程打断,只是把中断标志位设置为true
if (shouldParkAfterFailedAcquire(p, node)) {
interrupted |= this.parkAndCheckInterrupt();
}
}
} catch (Throwable var5) {
//没有捕捉中断异常...
this.cancelAcquire(node);
if (interrupted) {
selfInterrupt();
}

throw var5;
}
}

lock.lockInterruptibly();:默认是可打断的

1
2
3
4
5
6
7
8
9
10
public final void acquireInterruptibly(int arg) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
if (!this.tryAcquire(arg)) {
this.doAcquireInterruptibly(arg);
}

}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private void doAcquireInterruptibly(int arg) throws InterruptedException {
AbstractQueuedSynchronizer.Node node = this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE);

try {
AbstractQueuedSynchronizer.Node p;
do {
p = node.predecessor();
if (p == this.head && this.tryAcquire(arg)) {
this.setHead(node);
p.next = null;
return;
}
} while(!shouldParkAfterFailedAcquire(p, node) || !this.parkAndCheckInterrupt());
//如果有线程打断就会退出上面的循环,并抛出这个中断异常
throw new InterruptedException();
} catch (Throwable var4) {
this.cancelAcquire(node);
throw var4;
}
}

5.6 条件变量

每个条件变量就对应着一个等待队列,实现类是ConditionObject(是AbstractQueuedSynchronizer的内部类)

1
2
3
4
Condition waitSet1 = lock.newCondition();
waitSet1.await();
Condition waitSet2 = lock.newCondition();
waitSet2.await();

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
public final void await() throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
//将线程关联一个节点,并将它加入到这个阻塞队列中
AbstractQueuedSynchronizer.Node node = this.addConditionWaiter();
//将线程上所有的锁都释放掉,因为锁是可重入的,所以要全部释放
int savedState = AbstractQueuedSynchronizer.this.fullyRelease(node);
int interruptMode = 0;

while(!AbstractQueuedSynchronizer.this.isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = this.checkInterruptWhileWaiting(node)) != 0) {
break;
}
}

if (AbstractQueuedSynchronizer.this.acquireQueued(node, savedState) && interruptMode != -1) {
interruptMode = 1;
}

if (node.nextWaiter != null) {
this.unlinkCancelledWaiters();
}

if (interruptMode != 0) {
this.reportInterruptAfterWait(interruptMode);
}

}
}

signal()

1
2
3
4
5
6
7
8
9
10
11
12
public final void signal() {
if (!AbstractQueuedSynchronizer.this.isHeldExclusively()) {
throw new IllegalMonitorStateException();
} else {
//每次取的都是队列的第一个
AbstractQueuedSynchronizer.Node first = this.firstWaiter;
if (first != null) {
this.doSignal(first);
}

}
}

doSignal()

1
2
3
4
5
6
7
8
9
10
private void doSignal(AbstractQueuedSynchronizer.Node first) {
do {
if ((this.firstWaiter = first.nextWaiter) == null) {
this.lastWaiter = null;
}

first.nextWaiter = null;
} while(!AbstractQueuedSynchronizer.this.transferForSignal(first) && (first = this.firstWaiter) != null);

}


Java多线程-ReentrantLock
https://vickkkyz.fun/2022/03/24/Java/JUC/ReentrantLock/
作者
Vickkkyz
发布于
2022年3月24日
许可协议