zoukankan      html  css  js  c++  java
  • 多线程学习笔记三之ReentrantLock与AQS实现分析

    简介

      ReentrantLock是基于同步器AbstractQueuedSynchronizer(AQS)实现的独占式重入锁,支持公平锁、非公平锁(默认是非公平锁)、申请锁可响应中断以及限时获取锁等高级功能,分析ReentrantLock就离不开同步器AQS,关系图如下:

      在AQS中实现了如何获取锁和释放锁的模板方法,重入锁ReentrantLock实现时通过内部类继承Sync同步器AbstractQueuedSynchronizer。并调用同步器提供的模板方法,而这些模板方法将会调用ReentrantLock重写的方法,这是典型的模板方法设计模式。AQS实现同步器功能离不开三大基础组件:

    • 对共享资源同步状态进行原子性管理 ---> 利用CAS对同步状态进行更新
    • 线程的阻塞与唤醒 ---> 调用native方法
    • 等待队列的管理 ---> 维护FIFO队列

    AQS同步状态

      AQS中使用了一个int型的volatile变量来表示同步状态,线程在尝试获取锁的时候,就回去比较同步器同步状态state是否为0,为0,那么线程就拿到了锁并改变同步状态;不为0,说明有其他线程拿到了锁。AQS中提供了以下三个方法来访问或修改同步状态:

    	//AQS成员变量,同步状态
    	private volatile int state;
    
    	//获取当前同步状态
    	protected final int getState() {
            return state;
        }
    
    	//设置当前同步状态
        protected final void setState(int newState) {
            state = newState;
        }
    
    	//使用CAS设置当前状态,该方法能够保证状态设置的原子性
        protected final boolean compareAndSetState(int expect, int update) {
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    

    AQS同步队列

      当有多个线程竞争获取锁时,只有一个线程能获取到锁,那么这些没有获取到锁的线程就需要等待,等到线程把锁释放了再唤醒等待线程去获取锁,为了实现等待-唤醒机制,AQS提供了基于CLH队列(Craig, Landin,Hagersten)实现的等待队列,是一个先入先出的双向队列。同步队列是一个非阻塞的 FIFO 队列。也就是说往里面插入或移除一个节点的时候,在并发条件下不会阻塞,而是通过自旋锁和CAS保证节点插入和移除的原子性。

      AQS中的内部类Node是构建同步队列和等待队列(后面介绍Condition再介绍)的基础节点类,Node类部分源码如下:

        static final class Node {
    		//等待状态
            volatile int waitStatus;
    
    		//前驱结点
            volatile Node prev;
    	
    		//后继节点
            volatile Node next;
    		
    		//等待获取锁的线程
            volatile Thread thread;
    		
    		//condition队列的后继节点
            Node nextWaiter;      
        }
    

    关于节点Node的waitStatus,它反映的是节点中线程的等待状态,有如下取值:

    • CANCELLED,值为1,因为超时或中断,该线程已经被取消
    • SIGNAL,值为-1,线程的后继线程正/已被阻塞,当该线程release或cancel时要重新这个后继线程(unpark)
    • CONDITION,值为-2,表明该线程被处于条件队列,就是因为调用了Condition.await而被阻塞
    • PROPAGATE,值为-3,表示当前场景下后续的acquireShared能够得以执行
    • 等待状态的初始值为0,表示当前节点在sync队列中,等待着获取锁。

    ReentrantLock数据结构

      从关系图可以看出,ReentrantLock实现了Lock接口,内部类Sync是AQS的子类,Sync有两个子类FairSync(公平锁)和NonFairSync(非公平锁)。ReentrantLock只有一个成员变量sync,通过构造函数初始化,可以看到通过默认的构造函数构造的ReentrantLock是非公平锁。

        private final Sync sync;
    
        public ReentrantLock() {
            sync = new NonfairSync();
        }
    
    	public ReentrantLock(boolean fair) {
            sync = fair ? new FairSync() : new NonfairSync();
        }
    

    公平锁的获取

      ReentrantLock获取锁方法如下:

        public void lock() {
            sync.lock();
        }
    

    公平锁调用的是FairSync的lock方法:

        final void lock() {
            acquire(1);
        }
    

    acquire方法是AQS实现的方法,介绍一下参数的1的意思:AQS规定同步状态state,想要获得锁就去改变同步状态,就是把同步状态加1。acquire方法:

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

    获取锁的过程:

    1. 尝试获取锁。
    2. 尝试获取失败,将当前线程构成Node加入Sync队列。
    3. 再次尝试获取,若获取失败线程进入等待态,等待唤醒。

    tryAcquire(arg)

      公平锁尝试获取,在FairSync里实现,获取同步状态成功返回true,否则返回false

        protected final boolean tryAcquire(int acquires) {
    		//获取当前线程
            final Thread current = Thread.currentThread();
    		//获取同步状态
            int c = getState();
    		//同步状态为0,没有其他线程占据锁
            if (c == 0) {
    			//检测同步队列没有其他线程等待(确保公平性),如果没有获取锁就以CAS方式尝试改变同步状态
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
    				//设置锁的拥有者为当前线程
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
    		//同步状态不为0,检测是否是当前线程拥有锁
            else if (current == getExclusiveOwnerThread()) {
    			//当前线程拥有锁,直接更新同步状态,重入锁
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    
    • hasQueuedPredecessors()
        hasQueuedPredecessors是AQS中的方法,检测同步队列有没有等待获取锁的线程,保证公平性。
        public final boolean hasQueuedPredecessors() {
            //同步队列尾节点
            Node t = tail;
    		//同步队列头节点 
            Node h = head;
            Node s;
    		//h!=t 头节点和尾节点不同,说明同步队列不为空
    		//同步队列不为空,检测下一个等待获取锁的线程(h.next.thread)是不是当前线程
            return h != t &&
                ((s = h.next) == null || s.thread != Thread.currentThread());
        }
    
    • compareAndSetState(int expect, int update)
        compareAndSetState()在AQS中实现。compareAndSwapInt() 是sun.misc.Unsafe类中的一个native方法,如果当前状态值等于预期值,则以原子方式将同步状态设置为给定的更新值。
        protected final boolean compareAndSetState(int expect, int update) {
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    
    • setExclusiveOwnerThread(Thread thread) & getExclusiveOwnerThread()
        setExclusiveOwnerThread和getExclusiveOwnerThread都是AQS父类AbstractOwnableSynchronizer的方法,setExclusiveOwnerThread用于设置线程t为当前拥有独占锁的线程。getExclusiveOwnerThread用于获得当前占据独占锁的线程
        protected final void setExclusiveOwnerThread(Thread thread) {
            exclusiveOwnerThread = thread;
        }
    
        protected final Thread getExclusiveOwnerThread() {
            return exclusiveOwnerThread;
        }
    

    addWaiter(Node mode)

      addWaiter在AQS中实现,以当前线程构成节点加入到同步队列末尾,并返回这个节点Node。

        private Node addWaiter(Node mode) {
    		//以当前线程和给定模式构成节点Node
            Node node = new Node(Thread.currentThread(), mode);
            // 同步队列不为空,以CAS方式把当前线程加入到队列末尾
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
    		//队列为空,建立同步队列,再把当前线程加入同步队列
            enq(node);
            return node;
        }
    
    
    • compareAndSetTail(Node expect, Node update)
        compareAndSetTail是AQS中的方法,调用本地native方法,如果同步队列队尾是expect节点,就把update节点添加到队列末尾,这是一个原子操作。
        private final boolean compareAndSetTail(Node expect, Node update) {
            return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
        }
    
    • enq(final Node node)
        private Node enq(final Node node) {
            for (;;) {
                Node t = tail;
                if (t == null) { // Must initialize
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }
    
    

    acquireQueued(final Node node, int arg)

      如果当前线程的节点的前驱结点,就去尝试获取同步状态,如果不是或者获取失败根据waitStatus对同步队列进行清理:把waitStatus为CANCELLED从同步队列清除,修改错误的waitStatus,然后把线程堵塞,返回当前线程是否被中断。

        final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
    				//当前节点的前驱结点
                    final Node p = node.predecessor();
    				//前驱结点是head头节点,尝试获取同步状态
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    
    
    • shouldParkAfterFailedAcquire(Node pred, Node node)
        前驱结点不是head头节点或尝试获取同步状态失败以后,并不是马上把当前线程线程堵塞,还要检测同步队列前驱结点的状态,检查规则如下:
    1. 如果前驱节点状态为SIGNAL,表明当前节点需要被堵塞,此时则返回true。
    2. 如果前驱节点状态为CANCELLED(ws>0),说明前继节点已经被取消,则从后往前找到一个有效(非CANCELLED状态)的节点,并返回false;之后无限循环直到步骤1返回true,线程阻塞。
    3. 如果前驱节点状态为非SIGNAL、非CANCELLED,则CAS设置前驱节点的状态为SIGNAL,并返回false;之后无限循环直到步骤1返回true,线程阻塞。
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)            
                return true;
            if (ws > 0) {            
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {            
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }
    
    
    • parkAndCheckInterrupt()
        把当前线程堵塞并检查是否有中断。
        private final boolean parkAndCheckInterrupt() {
            LockSupport.park(this);
            return Thread.interrupted();
        }
    
    

    锁的释放

      ReentrantLock公平锁与非公平锁的释放机制是一样的,释放锁方法如下:

        public void unlock() {
            sync.release(1);
        }
    

    unlock方法调用的release方法是在AQS中实现的,这里的1类似于acquire(1),适用于用来设置同步状态的,释放锁时会把同步状态减1。release方法会先调用tryRelease来尝试释放当前线程锁持有的锁。成功的话,则唤醒后继等待线程,并返回true。否则,直接返回false

        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(int releases)

      tryRelease尝试获取锁,当同步状态为0时清空占据锁的线程,返回true;如果同步状态不为0返回false,因为ReentrantLock是重入锁,只有彻底释放tryRelease才会返回true。

        protected final boolean tryRelease(int releases) {
    		// c是本次释放锁之后的同步状态
            int c = getState() - releases;
    		//当前线程不是锁的拥有者,抛出IllegalMonitorStateException异常
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
    		//如果“锁”已经被当前线程彻底释放,则设置“锁”的持有者为null,即锁是可获取状态。
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
    

    unparkSuccessor(Node node)

      当前线程释放锁成功的话,会唤醒当前线程的后继线程。从aquireQueued方法可以看出,一旦头结点的后继结点被唤醒,那么后继结点就尝试去获取锁,如果获取成功就将头结点设置为自身,并将前一个头节点清空。

        private void unparkSuccessor(Node node) {
            // 获取当前线程(要释放锁)的等待状态
            int ws = node.waitStatus;
            if (ws < 0)
    			//设置为初始状态
                compareAndSetWaitStatus(node, ws, 0);
    
            //同步队列头节点的下一个等待节点
            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);
        }
    

    非公平锁的获取

      NonfairSync类中lock()实现,首先尝试用CAS更改同步状态,如果成功,把当前线程设置为独占锁的拥有者;然后调用acquire(1)方法。

        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
    

    acquire方法除了tryAcquire是由AQS的子类实现的,其他方法都是在AQS类实现的,tryAcquire的实现机制不同体现了公平锁与非公平锁的不同。

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

    ReentrantLock中的NonfairSync的tryAcquire方法,调用了nonfairTryAcquire方法

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

    nonfairTryAcquire(int acquires)

      非公平锁的尝试获取锁时,如果同步状态为0,即没有其他线程获取到锁,当前线程直接以CAS方式改变同步状态,不会去同步队列找是否有其他线程早于当前线程等在同步队列中,效率较高。

        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
    		//同步状态为0,尝试以CAS方式改变同步状态
            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");
                setState(nextc);
                return true;
            }
            return false;
        }
    

    总结

      本文介绍了ReentrantLock基于AQS同步器实现的公平锁和非公平锁的获取和释放,基于CAS改变同步状态是获得独占锁的基础,为了避免多个线程同时对进行竞争,在AQS中维护了FIFO的同步队列,当独占锁释放时,AQS同步器调度同步队列队首等待节点的线程去获取锁,有效避免了海量竞争独占锁造成资源的浪费,是一个非常巧妙的方法。

  • 相关阅读:
    Lambda 表达式
    jmeter--基于http+json接口的功能测试
    【python】-- web开发之jQuery
    【python】-- web开发之DOM
    【python】-- web开发之JavaScript
    【python】-- web开发之CSS
    【python】-- web开发之HTML
    【Robot Framework】---- Robot Framework简介、特点、RIDE
    【python】-- paramiko、跳板机(堡垒机)
    【python】-- pymsql 外键
  • 原文地址:https://www.cnblogs.com/rain4j/p/10075385.html
Copyright © 2011-2022 走看看