写点什么

万字长文!从底层开始带你了解并发编程,彻底帮你搞懂 Java 锁!

  • 2021 年 11 月 11 日
  • 本文字数:8006 字

    阅读完需:约 26 分钟

研究了AQS一天,终于找到了他的入口,接下来看我的想法:

多线程操做共享数据问题

/**


  • @author yhd

  • @createtime 2020/9/8 8:11*/public class Demo1 {public static int m=0;


public static void main(String[] args)throws Exception {Thread []threads=new Thread[100];for (int i = 0; i < threads.length; i++) {threads[i]=new Thread(()->{for (int j = 0; j < 100; j++) {m++;}});}for (Thread t :threads) t.start();for (Thread t :threads) t.join();//线程顺序结束 System.out.println(m);}}


毫无疑问,这段代码是存在线程安全问题的,只要了解一点并发编程,都是可以看出来的。那么我们可以怎么来解决呢?


使用 synchronized 来解决


/**


  • @author


【一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义】
浏览器打开:qq.cn.hn/FTf 免费领取
复制代码


yhd


  • @createtime 2020/9/8 8:32*/public class Demo2 {


public static int m=0;


public static void main(String[] args)throws Exception {Thread []threads=new Thread[100];for (int i = 0; i < threads.length; i++) {threads[i]=new Thread(()->{synchronized (Demo2.class) {for (int j = 0; j < 100; j++) {m++;}}});}for (Thread t :threads) t.start();for (Thread t :threads) t.join();//线程顺序结束 System.out.println(m);}


这样解决了线程安全的问题,但是同时也存在一个问题,synchronized 是操作系统层面的方法,也就是需要 jvm 和操做系统之间进行一个切换(用户态和内核态的切换),这样实际上是影响效率的。另一种解决办法:


使用 ReentrantLock 来解决


/**


  • @author yhd

  • @createtime 2020/9/8 8:41*/public class Demo3 {


public static int m=0;public static Lock lock=new ReentrantLock();public static void main(String[] args)throws Exception {Thread []threads=new Thread[100];for (int i = 0; i < threads.length; i++) {threads[i]=new Thread(()->{


try {lock.lock();for (int j = 0; j < 100; j++) {m++;}} finally {lock.unlock();}});}for (Thread t :threads) t.start();for (Thread t :threads) t.join();//线程顺序结束 System.out.println(m);}}


那么这种方式的底层是怎么做的呢?接下来追源码。


public ReentrantLock() {sync = new NonfairSync();}


这个 sync 又是啥呢?接着追


static final class NonfairSync extends Sync {private static final long serialVersionUID = 7316153563782823691L;


/**


  • Performs lock. Try immediate barge, backing up to normal

  • acquire on failure.*/final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}


protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}}


它实际上是ReentrantLock的一个内部类继承了Sync,而他里面的方法实际上也是调用了Sync的方法。这样目标就明确了,我们可以看一下Sync。 这个Sync的源码:


abstract static class Sync extends AbstractQueuedSynchronizer


由此可知,实际上ReentrantLock是利用AbstractQueuedSynchronizer也就是AQS来实现的。

AbstractQueuedSynchronizer 概述

这个类的内部有一个内部类Node


static final class Node {static final Node SHARED = new Node();volatile Node prev;//前驱指针 volatile Node next;//后继指针 volatile Thread thread;//当前线程 private transient volatile Node head;//头节点 private transient volatile Node tail;//尾节点 private volatile int state;//锁状态,加锁成功则为 1,重入+1 解锁则为 0.....}


看到这里就想到了LinkedHashMap,实际上也就是类似于维持了一个双向的链表,每个节点都是一个线程。



它维护了一个 volatile int state(代表共享资源)和一个 FIFO 线程等待队列(多线程争用资源被阻塞时会进入此队列)。这里 volatile 是核心关键词,具体 volatile 的语义,在此不述。state 的访问方式有三种:


getState()setState()compareAndSetState()


AQS定义两种资源共享方式:Exclusive(独占,只有一个线程能执行,如ReentrantLock)和Share(共享,多个线程可同时执行,如Semaphore/CountDownLatch)。   不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源 state 的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS 已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:


isHeldExclusively():该线程是否正在独占资源。只有用到 condition 才需要去实现它。tryAcquire(int):独占方式。尝试获取资源,成功则返回 true,失败则返回 false。tryRelease(int):独占方式。尝试释放资源,成功则返回 true,失败则返回 false。tryAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0 表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回 true,否则返回 false。


ReentrantLock为例,state 初始化为 0,表示未锁定状态。A 线程 lock()时,会调用tryAcquire()独占该锁并将 state+1。此后,其他线程再tryAcquire()时就会失败,直到 A 线程 unlock()到 state=0(即释放锁)为止,其它线程才有机会获取该锁。当然,释放锁之前,A 线程自己是可以重复获取此锁的(state 会累加),这就是可重入的概念。但要注意,获取多少次就要释放多么次,这样才能保证 state 是能回到零态的。


再以CountDownLatch以例,任务分为 N 个子线程去执行,state 也初始化为 N(注意 N 要与线程个数一致)。这 N 个子线程是并行执行的,每个子线程执行完后countDown()一次,state 会CAS减 1。等到所有子线程都执行完后(即 state=0),会unpark()主调用线程,然后主调用线程就会从 await()函数返回,继续后余动作。


一般来说,自定义同步器要么是独占方法,要么是共享方式,他们也只需实现tryAcquire-tryReleasetryAcquireShared-tryReleaseShared中的一种即可。但 AQS 也支持自定义同步器同时实现独占和共享两种方式,如ReentrantReadWriteLock

源码解读与执行流程分析

独占锁

acquire(int)== 此方法是独占模式下线程获取共享资源的顶层入口。如果获取到资源,线程直接返回,否则进入等待队列,直到获取到资源为止,且整个过程忽略中断的影响。这也正是 lock()的语义,当然不仅仅只限于 lock()。获取到资源后,线程就可以去执行其临界区代码了。


public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}


函数流程如下:


tryAcquire()尝试直接去获取资源,如果成功则直接返回; addWaiter()将该线程加入等待队列的尾部,并标记为独占模式; acquireQueued()使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回 true,否则返回 false。如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断 selfInterrupt(),将中断补上。 tryAcquire(int) 此方法尝试去获取独占资源。如果获取成功,则直接返回 true,否则直接返回 false。这也正是 tryLock()的语义,还是那句话,当然不仅仅只限于 tryLock()。


protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();}


一开始还傻乎乎的想为啥直接抛异常了,后来才反应过来,这不是给自定义的方法么?AQS 这里只定义了一个接口,具体资源的获取交由自定义同步器去实现了。 addWaiter(Node) 此方法用于将当前线程加入到等待队列的队尾,并返回当前线程所在的结点。


private Node addWaiter(Node mode) {//以给定模式构造结点。mode 有两种:EXCLUSIVE(独占)和 SHARED(共享)Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;}


Node 结点是对每一个访问同步代码的线程的封装,其包含了需要同步的线程本身以及线程的状态,如是否被阻塞,是否等待唤醒,是否已经被取消等。变量 waitStatus 则表示当前被封装成 Node 结点的等待状态,共有 4 种取值 CANCELLED、SIGNAL、CONDITION、PROPAGATE。


CANCELLED:值为 1,在同步队列中等待的线程等待超时或被中断,需要从同步队列中取消该 Node 的结点,其结点的 waitStatus 为 CANCELLED,即结束状态,进入该状态后的结点将不会再变化。


SIGNAL:值为-1,被标识为该等待唤醒状态的后继结点,当其前继结点的线程释放了同步锁或被取消,将会通知该后继结点的线程执行。说白了,就是处于唤醒状态,只要前继结点释放锁,就会通知标识为 SIGNAL 状态的后继结点的线程执行。


CONDITION:值为-2,与 Condition 相关,该标识的结点处于等待队列中,结点的线程等待在 Condition 上,当其他线程调用了 Condition 的 signal()方法后,CONDITION 状态的结点将从等待队列转移到同步队列中,等待获取同步锁。


PROPAGATE:值为-3,与共享模式相关,在共享模式中,该状态标识结点的线程处于可运行状态。


0状态:值为 0,代表初始化状态。


AQS 在判断状态时,通过用 waitStatus>0 表示取消状态,而 waitStatus<0 表示有效状态。


== enq(Node)== 此方法用于将 node 加入队尾。


private Node enq(final Node node) {for (;;) {//自旋 Node t = tail;if (t == null) { // Must initialize// 队列为空,创建一个空的标志结点作为 head 结点,并将 tail 也指向它 if (compareAndSetHead(new Node()))tail = head;} else {//正常放入队列尾部 node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}}


cas自旋volatile变量 acquireQueued(Node, int) 通过 tryAcquire()和 addWaiter(),该线程获取资源失败,已经被放入等待队列尾部了。聪明的你立刻应该能想到该线程下一步该干什么了吧:进入等待状态休息,直到其他线程彻底释放资源后唤醒自己,自己再拿到资源,然后就可以去干自己想干的事了。没错,就是这样!是不是跟医院排队拿号有点相似~~acquireQueued()就是干这件事:在等待队列中排队拿号(中间没其它事干可以休息),直到拿到号后再返回。这个函数非常关键,还是上源码吧:


final boolean acquireQueued(final Node node, int arg) {boolean failed = true;//标记是否已经拿到锁 try {boolean interrupted = false;//标记等待过程中是否被中断过 for (;;) {final Node p = node.predecessor();//拿到前驱节点//如果前驱是 head,即该结点已成老二,那么便有资格去尝试获取资源(可能是老大释放完资源唤醒自己的,当然也可能被 interrupt 了)。if (p == head && tryAcquire(arg)) {setHead(node);//拿到资源后,将 head 指向该结点。所以 head 所指的标杆结点,就是当前获取到资源的那个结点或 null。p.next = null; // help GCfailed = false;return interrupted;}//如果自己可以休息了,就进入 waiting 状态,直到被 unpark()if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;//如果等待过程中被中断过,哪怕只有那么一次,就将 interrupted 标记为 true}} finally {if (failed)cancelAcquire(node);}}


shouldParkAfterFailedAcquire(Node, Node) 此方法主要用于检查状态,看看自己是否真的可以去休息了。


private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;//拿到前驱的状态 if (ws == Node.SIGNAL)//如果已经告诉前驱拿完号后通知自己一下,那就可以安心休息了 return true;if (ws > 0) {/*


  • 如果前驱放弃了,那就一直往前找,直到找到最近一个正常等待的状态,并排在它的后边。

  • 注意:那些放弃的结点,由于被自己“加塞”到它们前边,它们相当于形成一个无引用链,稍后就会被保安大叔赶走了(GC 回收)!*/do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {//如果前驱正常,那就把前驱的状态设置成 SIGNAL,告诉它拿完号后通知自己一下。有可能失败,人家说不定刚刚释放完呢!compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}


整个流程中,如果前驱结点的状态不是 SIGNAL,那么自己就不能安心去休息,需要去找个安心的休息点,同时可以再尝试下看有没有机会轮到自己拿号。 parkAndCheckInterrupt() 如果线程找好安全休息点后,那就可以安心去休息了。此方法就是让线程去休息,真正进入等待状态。


private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}``


Thread.interrupted()会清除当前线程的中断标记位。==整个获取锁的流程:==1.如果尝试获取锁成功,直接返回。2.没成功,先加入到等待队列尾部,标记为独占模式。3.尝试这获取一次锁后,如果还是获取不到就去休息,有机会时(轮到自己,会被 unpark())会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过,则返回 true,否则返回 false。4.如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断 selfInterrupt(),将中断补上。这也就是 ReentrantLock.lock()的流程



==release(int)==此方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即 state=0),它会唤醒等待队列里的其他线程来获取资源。


public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);//唤醒等待队列里的下一个线程 return true;}return false;}


它是根据 tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自定义同步器在设计 tryRelease()的时候要明确这一点!! tryRelease(int) 此方法尝试去释放指定量的资源。


protected boolean tryRelease(int arg) {throw new UnsupportedOperationException();}


还是需要 AQS 的实现类自己去写。 unparkSuccessor(Node) 此方法用于唤醒等待队列中下一个线程。


private void unparkSuccessor(Node node) {//这里,node 一般为当前线程所在的结点。int ws = node.waitStatus;if (ws < 0)//置零当前线程所在的结点状态,允许失败。compareAndSetWaitStatus(node, ws, 0);


//找到下一个需要唤醒的结点 sNode s = node.next;if (s == null || s.waitStatus > 0) {//如果为空或已取消 s = null;for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)//从这里可以看出,<=0 的结点,都是还有效的结点。s = t;}if (s != null)LockSupport.unpark(s.thread);//唤醒}


用 unpark()唤醒等待队列中最前边的那个未放弃线程。 释放锁的流程 1.释放指定锁的资源并返回结果。 2.如果成功释放,就唤醒等待队列中最前边的那个未放弃线程。 3.如果没成功,返回 false。

共享锁

public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg);}


此方法用于将当前线程加入等待队列尾部休息,直到其他线程释放资源唤醒自己,自己成功拿到相应量的资源后才返回。


private void doAcquireShared(int arg) {final Node node = addWaiter(Node.SHARED);boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {//成功 setHeadAndPropagate(node, r);//将 head 指向自己,还有剩余资源可以再唤醒之后的线程 p.next = null; // help GCif (interrupted)selfInterrupt();failed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}


跟独占模式比,这里只有线程是 head.next 时(“老二”),才会去尝试获取资源,有剩余的话还会唤醒之后的队友。那么问题就来了,假如老大用完后释放了 5 个资源,而老二需要 6 个,老三需要 1 个,老四需要 2 个。老大先唤醒老二,老二一看资源不够,他是把资源让给老三呢,还是不让?答案是否定的!老二会继续 park()等待其他线程释放资源,也更不会去唤醒老三和老四了。独占模式,同一时刻只有一个线程去执行,这样做未尝不可;但共享模式下,多个线程是可以同时执行的,现在因为老二的资源需求量大,而把后面量小的老三和老四也都卡住了。当然,这并不是问题,只是 AQS 保证严格按照入队顺序唤醒罢了(保证公平,但降低了并发)。 == setHeadAndPropagate(Node, int)==


private void setHeadAndPropagate(Node node, int propagate) {Node h = head; // Record old head for check belowsetHead(node);if (propagate > 0 || h == null || h.waitStatus < 0 ||(h = head) == null || h.waitStatus < 0) {Node s = node.next;if (s == null || s.isShared())doReleaseShared();}}


== doReleaseShared()==


private void doReleaseShared() {for (;;) {Node h = head;if (h != null && h != tail) {int ws = h.waitStatus;if (ws == Node.SIGNAL) {if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue; // loop to recheck casesunparkSuccessor(h);}else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue; // loop on failed CAS}if (h == head) // loop if head changedbreak;}}

自定义锁

不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源 state 的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS 已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:


isHeldExclusively():该线程是否正在独占资源。只有用到 condition 才需要去实现它。tryAcquire(int):独占方式。尝试获取资源,成功则返回 true,失败则返回 false。tryRelease(int):独占方式。尝试释放资源,成功则返回 true,失败则返回 false。tryAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0 表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回 true,否则返回 false。


自定义一个简单的锁


/**


  • @author yhd

  • @createtime 2020/9/8 9:44*/public class MLock implements Lock {private AbstractQueuedSynchronizer sync=new Sync();@Overridepublic void lock() {sync.acquire(1);}


@Overridepublic void lockInterruptibly() throws InterruptedException {


}


@Overridepublic boolean tryLock() {return false;}


@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return false;}


@Overridepublic void unlock() {sync.release(1);}


@Overridepublic Condition newCondition() {return null;}


//自定义一个独占锁 private class Sync extends AbstractQueuedSynchronizer {@Overrideprotected boolean tryAcquire(int arg) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}


@Overrideprotected boolean tryRelease(int arg) {assert arg == 1;if (!isHeldExclusively())throw new IllegalMonitorStateException();setExclusiveOwnerThread(null);setState(0);return true;}


@Overrideprotected boolean isHeldExclusively() {return getExclusiveOwnerThread() == Thread.currentThread();}}


}


Demo 测试:


/**


  • @author yhd

  • @createtime 2020/9/8 9:36*/public class Demo6 {public static int m = 0;public static Lock lock = new MLock();

评论

发布
暂无评论
万字长文!从底层开始带你了解并发编程,彻底帮你搞懂Java锁!