`
碧海山城
  • 浏览: 189300 次
  • 性别: Icon_minigender_1
  • 来自: 杭州
社区版块
存档分类
最新评论

AQS:ReentrantLock源码分析

 
阅读更多

 

接着上一篇的

 

关于java线程(4)----JUC之 AQS 状态依赖的抽象

看一下ReentrantLock的源码,这里只是从AQS的角度出发,并不是从Lock的角度来看,那个以后再分析把

 

 

AQS的状态角度,代码整体结构上是这样的:

 

//检查状态
while(!checkAndChangeState){
	enque(currentthread)  //将当前线程加入等待队列
	park(currentThread)	//挂起当前线程
}
//do sth
…….

//释放锁,恢复状态
changeState(){
	Dequeue(current)	//出对
	Unpark(queue.thread)//唤醒其他等待中的线程
	
}

 

 和真正的代码比起来主要是一些小地方的优化,比如一些自旋操作,特别是对于很小范围内的锁,另外,就是队列中的线程可能是已经取消的,这也要做相应的处理,下面就具体分析下

 

Lock的结构体系成对的主要lockunlock其他方法到时候再看把:


 我们看下lock方法:

 

public void lock() {
        sync.lock();
}
//先暂且不管公平锁还是非公平锁,从独占锁的角度来说,就是请求一个状态
final void lock() {
            acquire(1);
}
 

看下acquire的代码,完全就是上面伪代码的结构

 

public final void acquire(int arg) {
		//判断状态是否ok(自己实现),不ok的话就加入队列,并且阻塞(父类实现)
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}

 

再来看下状态判断,这个是需要每个synchronizer自己做的:

 

final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
//这里初看起来会有并发问题,但是下面的cas操作保证了不会有并发问题,并且state字段是volatile的
            if (c == 0) { 
                if (compareAndSetState(0, acquires)) { 
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
	//重入锁,单个线程内,不会有并发性的,所以直接set就好
                setState(nextc);                 
	return true;
            }
            return false;
        }
 

这个方法如果返回true的话,代表条件允许,这样线程也就获得到锁了,其他啥也不用做了,等着释放锁,但是如果返回false,表示其他线程占用着,就需要加入等待队列,并且阻塞线程:

 

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

 

先看addWaiter

 

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // 这里假设队列非空,并且没有太多的并发性,尝试快速入队
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //队列为空,或者存在高并发,入队的时候失败了,需要用while尝试入队
	enq(node);     
        return node;
	}

   // 一个完整的入队操作,需要在一个循环里面判断队列是否为空、高并发等情况

 

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // 队列为空,需要初始化
                Node h = new Node(); // Dummy header
                h.next = node;
                node.prev = h;
                if (compareAndSetHead(h)) {//设置队列头,原子操作
                    tail = node;
                    return h;
                }
            }
            else {// 队列非空,直接插到最后面

                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

 

//在入队之后,在看下真正阻塞线程的acquireQueued,这里是一个循环操作,唤醒的时候也是重新去竞争,不是立刻获得锁
final boolean acquireQueued(final Node node, int arg) {
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
//在真正阻塞线程之前,如果发现自己前面的节点是head,那还要再尝试下去获取锁,获取到以后就把head踢了,自己当head,因为head的意义表示正在占用锁的节点(某些同步范围很短的锁很有用)                
if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
//该方法判断是否需要阻塞当前线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt()) //park当前线程
interrupted = true;        //默认不处理interrupted
            }
        } catch (RuntimeException ex) {
            cancelAcquire(node);   //发送异常的情况下,取消当前节点,并且如果当前节点是head,或者当前节点是等待唤醒状态,那么,还要尝试唤醒后面的节点
            throw ex;
        }
}

 

	//决定是否需要阻塞当前线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int s = pred.waitStatus;
//如果当前节点的前一个节点的状态小于0(signal/condition),表示前面那个节点也在等待唤醒,果断把自己挂起
        if (s < 0)
            return true;
//如果前面节点的状态大于0,(cancelled),表示前面的线程已经取消,向前遍历,直到找到状态不大于0的节点 
        if (s > 0) {        
	   		  do {
			 	node.prev = pred = pred.prev;
	  	  	 } while (pred.waitStatus > 0);
	  		 pred.next = node;
		} else 
			/*
	* 前面一个节点的状态为0,那么就是前一个节点认为他后面没有节点需要唤醒	* 啦,这时候要果断把他的状态改为SIGNAL,因为SIGNAL状态表示后面有阻塞      
	线程需要唤醒. 
             */
            compareAndSetWaitStatus(pred, 0, Node.SIGNAL);

	/**
	在state>=0的情况下,会走到这里,这里不能返回true把自己挂起,
	因为这时候线程切换,占用锁的线程A已经结束(并且发出了unpark信号),如果这时候线程B直接返回true阻塞自己,那可能会因为错失信号B永远无法唤醒;返回false,当前线程会去再次尝试获取锁,如果还是不能获取,则阻塞;
	*/
        return false;
	}
	//挂起当前线程,因为park会相应中断,但是不是抛出异常,因此这里将是否中断作为boolean类型返回,交给外部代码处理
	 private final boolean parkAndCheckInterrupt() {
	        LockSupport.park(this);
	        return Thread.interrupted();
	//并不是只有unpark和interrupt才能唤醒他,参考前面的关于LockSupport的讲解;唤醒的线程,重新去竞争锁,在高并发的情况下,有可能另一个节点正好也在请求lock,那么他刚好tryAcquire成功了,则当前线程又会重新阻塞!
	    }
 

 

 释放锁

 

/**
     *  释放锁主要做下面两件事情:
	 *	1.修改状态,改为可以获取锁
	 *  2.如果有等待的线程,则唤醒一个
     */
public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
			//如果waitStatus为0,则表示后面没阻塞线程了,没必要进行唤醒了
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
 

//先看修改状态的方法,这里只修改锁的状态,不修改队列的任何东西

protected final boolean tryRelease(int releases) {
     int c = getState() - releases;
     if (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) {
		/*
         * 把自己状态改为0,表示后面没有节点需要唤醒;如果后面有需要唤醒的节点,在请求锁的时候会把他的状态改为SIGNAL,不是很明白为什么这么做,难道是一种小小的优化?
         */
        int ws = node.waitStatus;
        if (ws < 0)
			compareAndSetWaitStatus(node, ws, 0);
        //取队列里面状态不大于0(cancelled)的节点,如果大于0,则从队列移除,然后从后往前找,找到最前面的一个非cancelled节点,并且唤醒这个节点
        Node 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)
                    s = t;
        }
		
        if (s != null)
            LockSupport.unpark(s.thread);
}
 

 

总结

1.通过AQSstate字段来表示锁是否被占用,特别需要注意的就是他的操作都是原子的

2.通过队列来保存阻塞的线程

3.通过LockSupport来进行挂起和唤醒操作(LockSupport的注意点可以看上一篇)

 

 

从挂起的时机上,AQS加了一些自旋操作,不是每次发现不能获取锁就挂起,而是后来会再次尝试获取锁,这样,对那些同步范围非常小,时间非常短的锁,应该是一种性能的提高。

 

另外,对于队列的操作,从我个人的角度来看,每次取当前正在跑的节点,跑完以后再取下一个节点,如果是取消状态,那么再去下一个,最后找到一个ok的唤醒,这样就可以了。不过AQS增加了一些状态判断,比如通过waitStatus来判断后面是否有节点需要唤醒,从角度来看,这完全增加了成本,不过性能提高多少就难说喽!

 

2
5
分享到:
评论
1 楼 zhongliangjun1 2012-12-14  
写得非常好,受益良多!

相关推荐

    7、深入理解AQS独占锁之ReentrantLock源码分析(1).pdf

    7、深入理解 AQS 独占锁之 Reentrantlock 源码分析 (1).pdf 8、读写锁ReentrantReadWriteLock&StampLock详解.pdf 9、并发容器 (Map、List、Set) 实战及其原理.pdf 10、阻塞队列BlockingQueue 实战及其原理分析.pdf

    Java并发包源码分析(JDK1.8)

    Java并发包源码分析(JDK1.8):囊括了java.util.concurrent包中大部分类的源码分析,其中涉及automic包,locks包(AbstractQueuedSynchronizer、ReentrantLock、ReentrantReadWriteLock、LockSupport等),queue...

    JUC AQS(AbstractQueuedSynchronizer)

    ReentrantLock Lock 加锁过程源码分析图,AQS 源码分析

    10、阻塞队列BlockingQueue实战及其原理分析.pdf

    7、深入理解 AQS 独占锁之 Reentrantlock 源码分析 (1).pdf 8、读写锁ReentrantReadWriteLock&StampLock详解.pdf 9、并发容器 (Map、List、Set) 实战及其原理.pdf 10、阻塞队列BlockingQueue 实战及其原理分析.pdf

    9、并发容器(Map、List、Set)实战及其原理.pdf

    7、深入理解 AQS 独占锁之 Reentrantlock 源码分析 (1).pdf 8、读写锁ReentrantReadWriteLock&StampLock详解.pdf 9、并发容器 (Map、List、Set) 实战及其原理.pdf 10、阻塞队列BlockingQueue 实战及其原理分析.pdf

    6、JUC并发工具类在大厂的应用场景详解(1).pdf

    7、深入理解 AQS 独占锁之 Reentrantlock 源码分析 (1).pdf 8、读写锁ReentrantReadWriteLock&StampLock详解.pdf 9、并发容器 (Map、List、Set) 实战及其原理.pdf 10、阻塞队列BlockingQueue 实战及其原理分析.pdf

    8、读写锁ReentrantReadWriteLock&StampLock详解.pdf

    7、深入理解 AQS 独占锁之 Reentrantlock 源码分析 (1).pdf 8、读写锁ReentrantReadWriteLock&StampLock详解.pdf 9、并发容器 (Map、List、Set) 实战及其原理.pdf 10、阻塞队列BlockingQueue 实战及其原理分析.pdf

    带你看看Java的锁(一)-ReentrantLock

    带你看看Javad的锁-ReentrantLock前言ReentrantLock简介Synchronized对比用法源码分析代码结构方法分析SyncNonfairSyncFairSync非公平锁VS公平锁什么是公平非公平ReentrantLockReentrantLock的构造函数lock加锁方法...

    Java并发编程学习笔记

    本文档主要内容如下: 1、线程安全和锁 Synchronized 底层实现原理 2、可重入锁与 Synchronized 的其他特性 3、ThreadLocal 的底层实现与使用 ...11、AQS源码分析 12、CAS原理分析和使用场景 13、.....

    java8源码-thread-concurrent-study:线程并发研究

    美团技术团队《从ReentrantLock的实现看AQS的原理及应用》 测试 老钱《打通Java任督二脉--并发数据结构的基石》 HongJie《一行一行源码分析清除AbstractQueuedSynchronizer》 爱吃鱼的KK ...

    带你看看Java的锁(二)-Semaphore

    带你看看Java的锁-Semaphore前言简介使用源码分析类结构图SyncNonfairSyncFairSyncSemaphore 构造函数Semaphore 成员方法获取释放总结 前言 简介 Semaphore 中文称信号量,它和ReentrantLock 有所区别,...

    leetcode下载-studyday:记得我

    源码分析 AQS同步器原理(实现各种同步器)模板方法模式,继承并冲重写AQS方法 主要是加锁和解锁 java 创建线程池相关 done 不推荐使用Excutors工具类去创建默认的几种线程池。会有OOM风险...要么核心线程树可能过多...

Global site tag (gtag.js) - Google Analytics