zoukankan      html  css  js  c++  java
  • ReentrantReadWriteLock源码解析

     public ReentrantReadWriteLock(boolean fair) {
            sync = fair ? new FairSync() : new NonfairSync();
            readerLock = new ReadLock(this);  
            writerLock = new WriteLock(this);
        }
    public final void acquireShared(int arg) {
            if (tryAcquireShared(arg) < 0)
                doAcquireShared(arg);
     
            static final int SHARED_SHIFT   = 16;
            static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
            static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
            static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
    
            /** Returns the number of shared holds represented in count  */
            static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }  
            /** Returns the number of exclusive holds represented in count  */
            static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

    c的低16位表示写锁重入的次数,高16位表示读线程的个数

    sharedCount表示高16位,exclusiveCounThread current = Thread.currentThread();

    int c = getState();
                if (exclusiveCount(c) != 0 &&
                    getExclusiveOwnerThread() != current) //如果有写线程且线程不是当前线程就返回-1
                    return -1;
                int r = sharedCount(c); 
                if (!readerShouldBlock() &&   //readershouldblock 如果阻塞队列里有写线程,读线程就不能获得锁
                    r < MAX_COUNT &&
                    compareAndSetState(c, c + SHARED_UNIT)) {  //通过cas操作将读线程的数量加1
                    if (r == 0) {  //r等于0说明没有读锁进入,
                        firstReader = current; 
                        firstReaderHoldCount = 1;
                    } else if (firstReader == current) {
                        firstReaderHoldCount++;
                    } else {
                        HoldCounter rh = cachedHoldCounter;
                        if (rh == null || rh.tid != getThreadId(current))
                            cachedHoldCounter = rh = readHolds.get(); //通过ThreadLocal给每个线程绑定了一个HoldCount对象,每个HoldCount都有一个count属性用来 
                                         //计数,还有一个ThreadId
    else if (rh.count == 0) readHolds.set(rh); rh.count++; } return 1; } return fullTryAcquireShared(current); }
    final int fullTryAcquireShared(Thread current) {
                /*
                 * This code is in part redundant with that in
                 * tryAcquireShared but is simpler overall by not
                 * complicating tryAcquireShared with interactions between
                 * retries and lazily reading hold counts.
                 */
                HoldCounter rh = null;
                for (;;) {                                                        //cas操作可能失败,失败的话,在循环获取读锁。
                    int c = getState();
                    if (exclusiveCount(c) != 0) {               //如果写线程不为0且线程不为当前线程就返回-1
                        if (getExclusiveOwnerThread() != current)
                            return -1;
                        // else we hold the exclusive lock; blocking here
                        // would cause deadlock.
                    } else if (readerShouldBlock()) {                           //即使readerShouldblock为true的情况下也可以获取读锁,因为可能这个线程之前
                        // Make sure we're not acquiring read lock reentrantly  //就获得过一次读锁,此次为重入
                        if (firstReader == current) {           //如果当前线程为第一个进入的读线程的话,count必然不为0
                            // assert firstReaderHoldCount > 0;
                        } else {                            
                            if (rh == null) {
                                rh = cachedHoldCounter;      
                                if (rh == null || rh.tid != getThreadId(current)) {
                                    rh = readHolds.get();
                                    if (rh.count == 0)      //判断当前线程获得过读锁,没获得过得话,count就为0,并remove掉HoldCount对象
                                        readHolds.remove();
                                }
                            }
                            if (rh.count == 0)
                                return -1;
                        }
                    }
                    if (sharedCount(c) == MAX_COUNT)
                        throw new Error("Maximum lock count exceeded");
                    if (compareAndSetState(c, c + SHARED_UNIT)) {
                        if (sharedCount(c) == 0) {
                            firstReader = current;
                            firstReaderHoldCount = 1;
                        } else if (firstReader == current) {
                            firstReaderHoldCount++;
                        } else {
                            if (rh == null)
                                rh = cachedHoldCounter;
                            if (rh == null || rh.tid != getThreadId(current))
                                rh = readHolds.get();
                            else if (rh.count == 0)
                                readHolds.set(rh);
                            rh.count++;
                            cachedHoldCounter = rh; // cache for release
                        }
                        return 1;
                    }
                }
            }

    如果tryAcquire返回-1的话就会进入到阻塞队列中去

    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);
                            p.next = null; // help GC
                            if (interrupted)
                                selfInterrupt();
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    private void setHeadAndPropagate(Node node, int propagate) {
            Node h = head; // Record old head for check below
            setHead(node);
            /*
             * Try to signal next queued node if:
             *   Propagation was indicated by caller,
             *     or was recorded (as h.waitStatus either before
             *     or after setHead) by a previous operation
             *     (note: this uses sign-check of waitStatus because
             *      PROPAGATE status may transition to SIGNAL.)
             * and
             *   The next node is waiting in shared mode,
             *     or we don't know, because it appears null
             *
             * The conservatism in both of these checks may cause
             * unnecessary wake-ups, but only when there are multiple
             * racing acquires/releases, so most need signals now or soon
             * anyway.
             */
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
                Node s = node.next;
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }

    读锁的unlock方法

    public void unlock() {
                sync.releaseShared(1);
            }
    public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }
     protected final boolean tryReleaseShared(int unused) {
                Thread current = Thread.currentThread(); 
                if (firstReader == current) {      //首先判断当前线程是否为第一个读线程
                    // assert firstReaderHoldCount > 0;
                    if (firstReaderHoldCount == 1) //在判断count数是否为0,为0的话,就说明读锁也完全释放
                        firstReader = null;  
                    else
                        firstReaderHoldCount--; //说明可重入锁还没有完全释放
                } else {
                    HoldCounter rh = cachedHoldCounter; //如果不是第一个线程的话,就通过ThreadLocal来获得HoldCount对象,
                    if (rh == null || rh.tid != getThreadId(current))
                        rh = readHolds.get();    
                    int count = rh.count;
                    if (count <= 1) {
                        readHolds.remove();
                        if (count <= 0)
                            throw unmatchedUnlockException();
                    }
                    --rh.count;
                }
                for (;;) {     //只释放一个读锁并不会唤醒阻塞队列里的线程,必须读锁和写锁都为0,即state为0才会唤醒.
                    int c = getState();
                    int nextc = c - SHARED_UNIT;
                    if (compareAndSetState(c, nextc))
                        // Releasing the read lock has no effect on readers,
                        // but it may allow waiting writers to proceed if
                        // both read and write locks are now free.
                        return nextc == 0;
                }
            }
    private void doReleaseShared() {
            /*
             * Ensure that a release propagates, even if there are other
             * in-progress acquires/releases.  This proceeds in the usual
             * way of trying to unparkSuccessor of head if it needs
             * signal. But if it does not, status is set to PROPAGATE to
             * ensure that upon release, propagation continues.
             * Additionally, we must loop in case a new node is added
             * while we are doing this. Also, unlike other uses of
             * unparkSuccessor, we need to know if CAS to reset status
             * fails, if so rechecking.
             */
            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 cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

    写锁的lock方法

     public void lock() {
                sync.acquire(1);
            }
    public final void acquire(int arg) {
            if (!tryAcquire(arg) &&                      //和ReentrantLock类似
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
    protected final boolean tryAcquire(int acquires) {
                /*
                 * Walkthrough:
                 * 1. If read count nonzero or write count nonzero
                 *    and owner is a different thread, fail.
                 * 2. If count would saturate, fail. (This can only
                 *    happen if count is already nonzero.)
                 * 3. Otherwise, this thread is eligible for lock if
                 *    it is either a reentrant acquire or
                 *    queue policy allows it. If so, update state
                 *    and set owner.
                 */
                Thread current = Thread.currentThread();
                int c = getState();
                int w = exclusiveCount(c);
                if (c != 0) {          //说明有锁,但不清楚是读锁还是写锁
                    // (Note: if c != 0 and w == 0 then shared count != 0)
                    if (w == 0 || current != getExclusiveOwnerThread()) //如果没有写锁或者有写锁但不是当前写锁都返回false。
                        return false;
                    if (w + exclusiveCount(acquires) > MAX_COUNT)
                        throw new Error("Maximum lock count exceeded");
                    // Reentrant acquire
                    setState(c + acquires); //走到这一步说明只有一个它本身的写锁,可以重入
                    return true;
                }
                if (writerShouldBlock() || //到了这一步就说明当前不含任何线程,cas操作state+1,并把线程设为系统线程
                    !compareAndSetState(c, c + acquires))
                    return false;
                setExclusiveOwnerThread(current);
                return true;
            }
    final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
                    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);
            }
        }

    和ReentrantLock的方法一样.

    写锁的unlock方法

      public void unlock() {
                sync.release(1);
            }
    public final boolean release(int arg) {
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    protected final boolean tryRelease(int releases) {
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                int nextc = getState() - releases;
                boolean free = exclusiveCount(nextc) == 0;
                if (free)
                    setExclusiveOwnerThread(null);
                setState(nextc);
                return free;
            }
    private void doReleaseShared() {
  • 相关阅读:
    Toolkit-常用函数库
    webpack 4版本以上 并行编译 thread-loader
    json格式数据下载
    npm 并行执行 concurrently
    WebAssembly assemblyscript
    npm outdated npm prune
    require import 引入区别
    C#数组的微软官方文档(链接)
    在C#中,override的成员仍然是virtual的
    SQL Server的DECIMAL类型,进行算术运算后会有精度变化
  • 原文地址:https://www.cnblogs.com/lzh66/p/13227521.html
Copyright © 2011-2022 走看看