多线程并发 (五) ReentrantLock 使用和源码
原创WangRain1 发布于2020-01-07 17:06:37 阅读数 110 收藏
章节：
- 多线程并发 (一) 了解 Java 虚拟机 - JVM
- 多线程并发 (二) 了解 Thread
- 多线程并发 (三) 锁 synchronized、volatile
- 多线程并发 (四) 了解原子类 AtomicXX 属性地址偏移量，CAS机制
- 多线程并发 (五) ReentrantLock 使用和源码
对于多线程并发学过了并发产生的原因，并发产生的问题，并发产生问题的解决方式，对于之前介绍的并发问题的解决方式有synchronzied、volatile、原子类型无锁控制。
了解最后一个锁ReentrantLock重入锁。ReentrantLock的实现其实是利用了CAS + volatile+LockSupport 的方式控制线程安全的，也就是面试经常问道，不用锁如何控制多线程安全。
1.ReentrantLock简单使用
ReentrantLock和synchronzied都是独占式重入锁，之前介绍过ReentrantLock是显示锁、synchronzied是内部锁，对于synchronzied的使用十分简单，能满足我们工作中的大部分需求。相对于ReentrantLock的使用就比synchronzied略有复杂，但是ReentrantLock能解决业务比较复杂的场景。
1) 对比
synchronzied锁的是对象（锁是保存在对象头里面的，根据对象头数据来标识是否有线程获得锁/争抢锁），ReentrantLock锁的是线程（根据进入的线程和int类型的state标识锁的获得/争抢）
synchronzied通过Object中的wait()/nofify()方法实现线程间通讯，ReentrantLock通过Condition的await()/signal()方法实现线程间通讯
synchronzied是非公平锁，ReentrantLock可选择公平锁/非公平锁
synchronzied涉及到锁的升级无锁->偏向锁->自旋锁->向OS申请重量级锁，ReentrantLock实现不涉及锁，利用CAS自旋机制和volatile同步队列实现锁的功能
ReentrantLock具有tryLock尝试获取锁以及tryLock timeout，可主动release释放使用灵活
2) 简单例子
public class Test {
static ReentrantLock lock = new ReentrantLock();
static Condition condition = lock.newCondition();
public static void main(String[] args) throws InterruptedException {
lock.lock();
new Thread(new SignalThread()).start();
System.out.println("等待通知");
try {
condition.await();
} finally {
lock.unlock();
}
System.out.println("恢复运行");
}
static class SignalThread implements Runnable {
@Override
public void run() {
lock.lock();
try {
condition.signal();
System.out.println("通知");
} finally {
lock.unlock();
}
}
}
}
使用了Condition + Reentrantlock实现线程间通信，和synchronzied的使用其实差别不大，使用的时候要保证lock()和unLock()方法的调用对应，调用次数保证相同。对于ReentrantLock的使用不做太多介绍，不熟悉的可以搜索用法。
2.ReentrantLock 源码实现
ReentrantLock其实是对 AbstractQueuedSynchronizer 子类 Sync 的一个封装，可以把ReentrantLock理解成一个包装类，主要逻辑都在AbstractQueuedSynchronizer（AQS） 和 Sync 子类里面，所以首先我们要学习的源码要从AQS开始。
代码结构图：
可知 ReentrantLock 分为公平锁FairSync和非公平锁NofairSync，这两种锁都是继承自Sync，并且是AQS的子类。
学习源码我们从两方面入手：1.数据结构、2.算法代码
AQS的数据结构
AQS是一个同步队列，是以Node类为一个节点的双向链表并且有首和尾指针。
// 首指针
private transient volatile Node head;
// 尾指针
private transient volatile Node tail;
// 是否有线程占用：0-无，1-有线程占用，>1-当前线程重入的次数
private volatile int state;
AQS中主要有三个参数而且都是被volatile修饰的，其中他们的更新方式是通过CAS机制Unsafe更新的，这块可以看多线程并发 (四) 了解原子类 AtomicXX 属性地址偏移量，CAS机制 了解CAS的参数含义。
Node内部类：
static final class Node {
volatile int waitStatus; //当前线程的等待状态
volatile Node prev;
volatile Node next;
volatile Thread thread; //当前线程
}
1）prev:指向前一个结点的指针
2) next:指向后一个结点的指针
3) thread:当前结点表示的线程,因为同步队列中的结点内部封装了之前竞争锁失败的线程,故而结点内部必然有一个对应线 程实例的引用
4) waitStatus:对于重入锁而言,主要有3个值。
0:初始化状态；
-1(SIGNAL):当前结点表示的线程在释放锁后需要唤醒后续节点的线程;
1(CANCELLED):在同步队列中等待的线程等待超时或者被中断,取消继续等待
1）队列中每个Node节点就代表一个等待获取锁的线程，其中head指的那个node节点就是当前当用锁的节点，当n1释放锁之后就会唤醒n2一次类推
2）当有新线程n3加入队列时候，就会从tail尾部加入，改变tail的指向。
从上图容易知道队列的结构，具体如何被添加进入队列又是如何释放的，继续看算法～
算法
1）线程被加锁/加入队列
从简单使用引入
public static void main(String[] args) throws InterruptedException {
ReentrantLock lock = new ReentrantLock();//初始化锁类型
lock.lock(); //进入加锁流程
try {
} finally {
lock.unlock(); //释放锁
}
}
初始化时候不传参数就是非公平锁，传参数跟据参数类型判断
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
主要以 NonfairSync 非公平锁代码为例，当调用lock()方法之后进入加锁流程
final void lock() {
if (compareAndSetState(0, 1)) //判断是否有线程获取了锁
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
1）compareAndSetState(0, 1) 利用CAS机制判断state属性是否被其他线程修改了，state = 0未被其他线程占用，state > 1被其他线程占用了
2）如果没被其他线程占用即state = 0,这时把当前线程设置到 AbstractOwnableSynchronizer 内存表示当前占用的线程
3）如果state != 0 ，继续 acquire(1) 把当前线程加入等待队列
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
1）首先会 tryAcquire(arg) 这个方法子类必须实现会引用到
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState(); // 当前状态
if (c == 0) { // 非公平的这里会再次尝试获取锁的机会和上面类似
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}//如果还是当前的线程说明当前线程重入了这个锁，state +1
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false; // 不是当前线程并且锁被其他线程占用了 返回false
}
可以看到这个方法主要分 三部分
1）第一部分if 中如果state=0了，那就直接占用这个锁，这里也是非公平锁的体现，并没有从队列中取，直接把锁让给了当前申请的线程
2）第二部分else if 中如果还是当前的线程那state +1 ，表示当前线程重入了这个锁
3）三 是个新的线程进入并且锁被其他线程占用，返回false
所以回到上面 当tryAcquire(arg) 返回true 结束，返回false继续走
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
这里涉及到了两个方法，首先是加入队列然后从队列中取出线程。主要逻辑处
首先看addWaiter()方法
private Node addWaiter(Node mode) {
Node node = new Node(mode); //创建一个新节点，mode = null
for (;;) { //无线循环
Node oldTail = tail; //拿到当前的未队列，
if (oldTail != null) { //不为空
U.putObject(node, Node.PREV, oldTail);
if (compareAndSetTail(oldTail, node)) { //移动尾部指针对象
oldTail.next = node; //把当前node加入队列
return node;
}
} else {
initializeSyncQueue(); //为空初始化 看下方
}
}
}
private final void initializeSyncQueue() {
Node h;
if (U.compareAndSwapObject(this, HEAD, null, (h = new Node()))) //给head赋值
tail = h; //给tail赋值
}
这块代码比较简单，主要说一下for循环中 oldTail !=null的那块
1）U.putObject(node, Node.PREV, oldTail); 这个是Unsafe 中的方法，意思是把oldTail 赋值给node中的 prev。
2）compareAndSetTail(oldTail, node) if 判断中的这块代码，意思是把tail这个指针从之前的oldTail指向node 看图
例如之前 tail = n2（oldTail） ，现在加入了一个线程n3,这时候 tail = n3
3）oldTail.next = node; 看图就是 n2.next = n3
继续回到acquireQueued()方法
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
通过addWaiter方法现在队列中已经加入了一个新的node节点。继续看acquireQueued方法
final boolean acquireQueued(final Node node, int arg) {
try {
boolean interrupted = false;
for (;;) { // 死循环
final Node p = node.predecessor(); //获取当前节点的上一个节点
if (p == head && tryAcquire(arg)) { //判断是否是head节点
setHead(node); // 把当前节点设置成head
p.next = null; // 把之前的head节点从链表中释放，让内存回收
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt()) // 暂停当前线程
interrupted = true;
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
这块代码比较好理解，我们传如的node是addwaiter方法return回来的，就是我们链表中最后一个节点，for循环中先通过node拿到最后一个节点的上一个结点，和head 首节点做比较，相同继续让当前线程 tryAcquire 获取当前锁，如果成功了那就是说上一个节点已经把锁释放了，当前节点就是链表中唯一一个节点了，然后把之前的节点p从链表中移除等待gc回收。如果获取没有成功判断是否需要暂停当前线程，如果pre节点的线程为SIGNAL状态那就调用LockSupport暂停当前线程。不然就一直循环直到前一个节点是head节点并且释放了锁。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 暂停当前线程
return Thread.interrupted();
}
在这里主要把线程暂停，因为当前node的前一个node释放锁之后会通知他。
2）线程释放锁/从队列中移除
释放锁相对简单通过主动调用unLock()方法，
public final boolean release(int arg) {
if (tryRelease(arg)) { //释放 state = 0
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h); // 解除线程的park等待
return true;
}
return false;
}
先是释放state的值，因为他是锁是否被占用的标识。然后unpark线程。
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
node.compareAndSetWaitStatus(ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
Node s = node.next; // 把取消的线程移除，轮寻直到线程没有被取消
if (s == null || s.waitStatus > 0) {
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread); //释放当前节点的线程
}
好的博文：
ReentrantLock应用
ReentrantLock源码
————————————————
版权声明：本文为CSDN博主「WangRain1」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/WangRain1/article/details/103869222