java thread 之Lock

java

concurrent包里面有很多Lock的具体实现,其具体的实现都是基于AQS实现的

ReentrantLock
ReentrantLock是可重入的互斥锁,重点是重入和互斥,ReentrantLock 将由最近成功获得锁的线程所持有,当这个线程再次尝试拥有这个Lock时就是重入。互斥就是
在某一时间只有一个线程能持有Lock。
    public void lock() {
        sync.lock();
    }
获得锁方法,Sync是AQS的抽象子类,实现可重入和互斥的大部分功能。在Sync的子类中有FairSync和NonfairSync两种代表公平锁策略和非公平锁策略

Sync lock方法留给子类去实现,NonfairSync的实现:
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
不管是不是有其他线程在AQS的阻塞的队列里面,如果当前线程能获得线程,就直接获得线程,不行的话执行AQS的acquire()方法,基本就是进入阻塞队列的命了。
关键点:AQS的acquire()中调用的tryAcquire(int arg)是留给子类实现的,是在进入阻塞队列前再尝试一次获取锁(lock()方法到这个点上面可能其他线程已经释放锁)

NonfairSync的tryAcquire(int arg)调用的nonfairTryAcquire()实现:
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {//关键点1
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;//关键点2
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
关键点1:state为0,没有线程请求锁,直接分配给本线程
关键点2.  如果本线程已经得到锁,state加1,即重入。

FairSync的实现:
    final void lock() {
            acquire(1);//关键点1
        }
    protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (isFirst(current) && //关键点2
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
关键点1:公平策略,没有给当前线程优惠
关键点2:判断当前线程有没有在阻塞队列的第一位,没在第一位不往下继续执行,这样先给阻塞队列的线程机会,这样做速度比较慢,吞吐量比较小

    public boolean tryLock() {
        return sync.nonfairTryAcquire(1);
    }
 tryLock尝试获得锁,不能获得话就直接返回,没有公平策略一说。

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;
    }
tryRelease()留给AQS子类执行:
        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;
        }
如果state被减后还不为0,表示这个锁被一个线程重入后还没有完全释放,release()方法后面不会执行,也就是不会执行unpark阻塞队列中的线程.


ReentrantReadWriteLock
ReentrantReadWriteLock内部有两个锁:互斥可重入的WriteLock和共享的ReadLock
内部实现中state字段的高16位代表的是read count,低16位代表的是write count
AQS的实现也有NonfairSync和FairSync之分:
主要区别是同时有读写线程时候对待读写线程策略不同。
NonfairSync:
final boolean writerShouldBlock(Thread current) {
            return false; // writers can always barge
        }
        final boolean readerShouldBlock(Thread current) {
            /* As a heuristic to avoid indefinite writer starvation,
             * block if the thread that momentarily appears to be head
             * of queue, if one exists, is a waiting writer. This is
             * only a probablistic effect since a new reader will not
             * block if there is a waiting writer behind other enabled
             * readers that have not yet drained from the queue.
             */
            return apparentlyFirstQueuedIsExclusive();
        }
write永远都不阻塞,read的话看阻塞队列header之后的node是不是write的,如果是就阻塞,可以避免read一直运行导致的write饥饿。
FairSync:
      final boolean writerShouldBlock(Thread current) {
            // only proceed if queue is empty or current thread at head
            return !isFirst(current);
        }
        final boolean readerShouldBlock(Thread current) {
            // only proceed if queue is empty or current thread at head
            return !isFirst(current);
        }
write和read都有判断是不是阻塞队列header后的第一个node。

对于ReadLock
     public void lock() {
            sync.acquireShared(1);
        }
调用AQS的acquireShared(),其中会执行留给AQS子类实现的tryAcquireShared():
 protected final int tryAcquireShared(int unused) {
            /*
             * Walkthrough:
             * 1. If write lock held by another thread, fail
             * 2. If count saturated, throw error
             * 3. Otherwise, this thread is eligible for
             *    lock wrt state, so ask if it should block
             *    because of queue policy. If not, try
             *    to grant by CASing state and updating count.
             *    Note that step does not check for reentrant
             *    acquires, which is postponed to full version
             *    to avoid having to check hold count in
             *    the more typical non-reentrant case.
             * 4. If step 3 fails either because thread
             *    apparently not eligible or CAS fails,
             *    chain to version with full retry loop.
             */
            Thread current = Thread.currentThread();
            int c = getState();
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            if (!readerShouldBlock(current) &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != current.getId())
                    cachedHoldCounter = rh = readHolds.get();
                rh.count++;
                return 1;
            }
            return fullTryAcquireShared(current);
        }
策略是以下的几点:
1.如果当前write持有锁,直接返回-1.
2.没有write持有锁,判断是否需要阻塞读请求,之后原子更新read count
3.上面成功后就返回1,不成功的话在循环中试着读取:
     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 = cachedHoldCounter;
            if (rh == null || rh.tid != current.getId())
                rh = readHolds.get();
            for (;;) {
                int c = getState();
                int w = exclusiveCount(c);
                if ((w != 0 && getExclusiveOwnerThread() != current) ||
                    ((rh.count | w) == 0 && readerShouldBlock(current)))
                    return -1;
                if (sharedCount(c) == MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                if (compareAndSetState(c, c + SHARED_UNIT)) {
                    cachedHoldCounter = rh; // cache for release
                    rh.count++;
                    return 1;
                }
            }
        }
unLock方法就是将read count 减1,然后查看阻塞队列有没有线程需要unpark

WriteLock的实现和ReetrantLock的实现是一致的
Semaphore
Semaphore是信号量的概念,主要控制同一时间内对访问资源的线程数的控制。底层实现是AQS的
acquireShared()和releaseShared().

CountDownLatch
CountDownLatch在完成一组正在其他线程中执行的操作之前,允许线程等待直到他们完成操作
await()方法一直等到state=0,初始时CountDownLatch初始化state为传入的数值,一个线程
执行操作完成的话执行countDown()方法将state减1,直到为0,await()方法不再阻塞。

CyclicBarrier
当一组线程调用CyclicBarrier.await()方法,就会阻塞在那里,当最后一个线程进入调用此方法时,就执行
一个公共的Runnable.原理是:初始化时有count属性,当前面count-1个线程到来都condition.await(),第count个
线程来就执行公共Runnable,然后执行condition.signAll()方法来使得其他线程从阻塞中解脱出来.

以上是 java thread 之Lock 的全部内容, 来源链接: utcz.com/z/393322.html

回到顶部