多线程四-Lock锁及其原理分析

news/2024/11/16 5:41:44/文章来源:https://www.cnblogs.com/caozz/p/18383738/thread0004

JUC是什么

可能有些不太关注底层代码，会不太理解juc是啥，比如之前的我，只知道是跟并发相关。juc其实就是并发包路径的缩写，java.util.concurrent.而Lock是其中锁的接口，有比如重入锁，读锁，写锁等一些具体实现。
这部分源码理解起来还是有些难度，暂时先理解其大概思路，对于实现有一个印象，比如AQS队列是一个双端队列，那么看代码时遇到相关操作知道是操作双端队列就容易一些了。
个人看新版本jdk的额实现，感觉更不容易理解，可能是因为jdk8网上的文章多一些，可以帮助理解吧。后面有时间还会对这部分进行重新思考。

Lock

lock的基本用法如下

Lock l = ...;  l.lock();
try {    // access the resource protected by this lock  
} finally {l.unlock();  
}

源码如下

trylock不阻塞，如果发现有人持有锁，则不去执行相应逻辑
lock则会阻塞在lock方法内部

package java.util.concurrent.locks;import java.util.concurrent.TimeUnit;public interface Lock {void lock();void lockInterruptibly() throws InterruptedException;boolean tryLock();boolean tryLock(long time, TimeUnit unit) throws InterruptedException;void unlock();Condition newCondition();
}

ReentrantLock和syncronized有什么区别

Syncronized是一个关键字，Lock是一个类
Lock锁的获取和释放需要手动操作,syncronized是自动的
syncronized是JVM层面的，Lock是API层面的
syncronized非公平，Lock可以通过参数设置
syncronized锁的是对象，锁信息保存在对象头中，而Lock是通过state来标识锁的状态
syncronized底层有一个锁升级的过程
都是来解决线程安全问题的

ReentrantLock

重入锁，互斥锁，基本Syncronized都可以用来代替
重入锁是用来解决死锁问题的。

package com.caozz.demo2.thread.concurrent;import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;public class LockExample {static Lock lock=new ReentrantLock();  //取决于这个对象实例的范围(锁的范围)private static int count=0;public static void inc(){lock.lock();// 抢占锁 //如果没有抢占到锁，会阻塞，阻塞在lock方法里面try {Thread.sleep(1);count++;} catch (InterruptedException e) {e.printStackTrace();}finally {lock.unlock();//}}public static void main(String[] args) throws InterruptedException {for (int i = 0; i < 1000; i++) {new Thread(()->LockExample.inc(),"t-"+i).start();}Thread.sleep(3000);System.out.println("result："+count);}
}

ReentrantLock的实现原理

满足互斥特性
意味着同一时刻，只允许一个线程进入到加锁的代码中
多线程环境下，线程的顺序访问。
猜测内部会有一个排号系统，来实现访问的顺序

锁的设计猜想

一定会涉及到锁的抢占，需要一个标记来实现互斥，全局变量
- 抢占到了锁，怎么处理？
  - 不需要处理，直接返回即可
- 没有抢占到，怎么处理
  - 需要等待(让处于排队中的线程，如果没有抢占到锁，则直接先阻塞-->释放CPU资源)
    - 如何让线程等待？
      - wait/notify
      - LockSupport.park/unpark
      - Condition
  - 需要排队(允许有N个线程被阻塞，此时线程处于活跃状态)
    - 通过一个数据结构，把这N个排队的线程存储起来
- 抢占到的锁，释放怎么处理
  - LockSupport.unpark() --> 唤醒处于队列中的制定线程
锁抢占的公平性（是否允许插队）
- 公平
- 非公平

锁竞争的实现原理

private volatile int state;状态，或者重入次数，可为0，或者大于0（可能会重入多次）。

公平锁与非公平锁，体现的地方在于，当ThreadA 执行完释放锁后，会把state状态改为0，然后去唤醒队列里面的线程。而在改完状态之后，如果ThreadD 来了，那么公平锁的话，D会加入到队列，非公平锁，则D直接除参与竞争锁。此时可能D会竞争到锁导致B抢不到锁。

AQS

state （锁的状态 0代表未加锁，>0则代表已加锁，和重入次数）
exclusiveOwnerThread （拥有当前锁的线程）

公平锁与非公平锁

通过ReentrantLock 构造函数，可以看出其默认是非公平锁

  //Sync里面的方法，继承了AQS，有实现类FairSync，NonFairSync分别用来实现公平锁与非公平锁public void lock() {sync.lock();   }//AQS里面的方法public final void acquire(int arg) {if (!tryAcquire(arg) && //抢占锁失败，就去进行排队//加入队列并进行自旋等待acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

公平锁

        final void lock() {acquire(1);}//抢占一个标记，成功返回true，失败返回false//只有AQS队列没有排队的线程，才会去竞争，这是与非公平锁的区别protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();//为什么不直接在state为0直接改状态？因为操作不是原子；为什么还要判断状态？双检锁if (c == 0) {                             //表示无锁状态if (!hasQueuedPredecessors() &&      //是否存在AQS队列，不存在才会去竞争锁compareAndSetState(0, acquires)) {   //cas原子操作（底层也是通过加锁保证线程安全的）setExclusiveOwnerThread(current);    //把获得锁的线程保存到exclusiveOwnerThread return true;}}//如果当前获得锁的线程和当前抢占锁的线程是同一个，则表示重入//此时当前线程已经获得锁，所以不用考虑并发else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;  // 增加重入次数if (nextc < 0)throw new Error("Maximum lock count exceeded");setState(nextc);    // 保存Statereturn true;}return false;}

非公平锁

        final void lock() {//不管当前AQS队列中是否有排队的情况先去插队竞争//在前一个线程刚刚释放，队列的线程准备去抢占的临界点，非公平锁在此时竞争if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());else    //即使抢占失败，调用tryAcquire->nonfairTryAcquire时，一样没有做是否有线程等待，也是直接去竞争acquire(1);}protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}

tryAcquire 抢占标记
公平锁的抢占多了一个条件， !hasQueuedPredecessors()

加入队列并进行自旋等待

acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

addWaiter(Node.EXCLUSIVE) //添加一个互斥锁的节点
acquireQueued() //自旋锁和阻塞的操作

    private Node addWaiter(Node mode) {//把当前线程封装成一个Node节点//后续唤醒线程的时候，需要得到被唤醒的线程Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failure//逻辑与enq类似，相当于尝试一次，如果未成功由enq来一直尝试Node pred = tail;if (pred != null) {node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;}//通过尾插法，入队//head与tail为全局变量，表示队列的头尾节点private Node enq(final Node node) {for (;;) {    //自旋Node t = tail;//刚初始化的队列为空，tail尾节点为空if (t == null) { // Must initializeif (compareAndSetHead(new Node()))tail = head;} else {//此时队列有值，t为原本的尾节点，node是这次要添加为尾节点的新的尾节点//将新的节点的前一个节点设置为尾节点node.prev = t;//将新节点node设置为新的尾节点if (compareAndSetTail(t, node)) {//将老的尾节点的下一个节点设置为新的尾节点nodet.next = node;return t;}}}}//node表示当前来抢占锁的线程，可鞥是队列里的任一线程final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {// 自旋//begin   ->尝试去获取锁，如果到 end结束，就相当于syncronized的轻量级锁，在不断的自旋final Node p = node.predecessor();  //当前节点的前置节点//此时也调用了tryAcquire，非公平锁在这里也会插队if (p == head && tryAcquire(arg)) {//如果返回true，则不需要等待，直接返回setHead(node);p.next = null; // help GCfailed = false;return interrupted;}// end//否则，让线程去阻塞if (shouldParkAfterFailedAcquire(p, node) &&  //shouldParkAfterFailedAcquire是否应该在抢占失败后阻塞parkAndCheckInterrupt())  //通过LockSupport.park来阻塞interrupted = true;}} finally {if (failed)cancelAcquire(node);}}/** waitStatus value to indicate thread has cancelled */static final int CANCELLED =  1;/** waitStatus value to indicate successor's thread needs unparking */static final int SIGNAL    = -1;/** waitStatus value to indicate thread is waiting on condition */static final int CONDITION = -2;/*** waitStatus value to indicate the next acquireShared should* unconditionally propagate*/static final int PROPAGATE = -3;private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;//默认节点是0if (ws == Node.SIGNAL)  //可以安全的阻塞return true;if (ws > 0) {  //只有一个CANCELLED ，不需要抢占锁，所以通过循环，将这些节点丢弃do {node.prev = pred = pred.prev;//从后往前清理} while (pred.waitStatus > 0);pred.next = node;} else {compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}

锁的释放

unlock方法，核心方法为unparkSuccessor(),为AQS里面的方法
先去修改状态，因为有可能存在重入

    public void unlock() {sync.release(1);}//AQS 方法public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}//ReentrantLock的方法protected final boolean tryRelease(int releases) {int c = getState() - releases;//重入情况下，可能还是大于0if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;if (c == 0) {free = true;setExclusiveOwnerThread(null);}//释放锁的过程，线程安全，因为还持有锁setState(c);return free;}private void unparkSuccessor(Node node) {int ws = node.waitStatus;if (ws < 0)   //表示可以唤醒状态，-2和-3与互斥锁无关compareAndSetWaitStatus(node, ws, 0);   //恢复成0Node s = node.next;if (s == null || s.waitStatus > 0) {//说明线程已经被销毁或出现异常，故将节点移除s = null;//从tail -> head进行遍历，因为在构建的时候是从prev往前，如果从head开始可能会造成队列不连续for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0) //只修改了prev节点状态，未修改自己，所以最后一个节点状态为0s = t;}if (s != null)LockSupport.unpark(s.thread);  //封装在Node中被阻塞的线程}