JDK中AbstractQueuedSynchronizer应用解析

Posted 2020-09-30 atheva

这个类首先是一个抽象类，定义了一个模板，很多java同步相关的类（ReetrantLock、Semaphore、CountDownLatch等)都是基于AbstractQueuedSynchronizer来实现的

AbstractQueuedSynchronizer

本身就是一个链表，提供的线程安全的操作。核心思想是通过CAS插入链表的尾部和获取链表的头结点。算法暂时先不谈，先说说他的应用，主要下先说说他的一些模板方法。

AbstractQueuedSynchronizer#acquire()

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

其中tryAcquire()就是一个模板方法，当tryAcquire()返回false的时候，就会把当前线程封装为Node(链表的数据结构)，然后挂起(使用LockSuport.park())当前线程。

其实在从队列中获取头节点的时候并恢复的时候，也会调用tryAcquire()方法，因为对一些非公平锁可能被强占。

AbstractQueuedSynchronizer#release()

和acquire()对应的就是release()

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()返回true的话，就会从链表中获取头节点（如果有)，并恢复线程。 因此，java并发包里很多都会用到这个来进行自定义。下面说举一个例子说说明如何利用上面2个模板方法来实现并发的

ReentrantLock

ReentrantLock的并发控制是通过继承AbstractQueuedSynchronizer()来实现并发的。

abstract static class Sync extends AbstractQueuedSynchronizer{
    ...
}

　　

而公平锁和非公平锁又是继承了Sync， 以默认的非公平锁分析，首先来tryAcquire()，通过注释可以看出基本步骤，如果要实现公平锁也很简单，在当前线程没有占有的情况下，多增加一个判断链表是否有等待的线程即可

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

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    // 0 表示当前没有其他线程占用
    if (c == 0) {
        //通过cas的方式来设置state,如果设置成功，则认为当前线程可以获取锁，cas失败则任务其他线程比他快一步占用
        if (compareAndSetState(0, acquires)) {
            //设置当前线程独占
            setExclusiveOwnerThread(current);
            //返回true,即不需要放入等待的链表中
            return true;
        }
    }
    //如果当前线程本来就占有锁
    else if (current == getExclusiveOwnerThread()) {
        //占用的次数+1,这也是为什么 lock()和unlock()需要一一对应
        int nextc = c + acquires;
        //说明同一个线程lock()的次数为int的最大值
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        // 设置state,返回获取锁成功
        setState(nextc);
        return true;
    }
    return false;
}

　　

再来看看tryRelease()

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的时候，设置独占锁为null并返回true即可。代码很简单。

小结

有人看到这里觉得，ReentrantLock是通过lock()和unlock()方法来加锁解锁的，和你说的acquire()和release()有什么关系呢？其实lock()的核心就是调用acquire(1),unlock()就是调用release(1)

AbstractQueuedSynchronizer#acquireShared()

acquireShared()和acquire()类似，只不过在调用的是

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

　　

tryAcquireShared()是模板方法，可以简单的理解为是把boolean类型换位int类型的acquire()

AbstractQueuedSynchronizer#releaseShared()

和acquireShared()对应acquire(),releaseShared()就是对应release()，就不在赘述

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

　　

下面再举个例子说明acquireShared()和releaseShared()的使用

Semaphore

Semaphore就是允许指定数量的线程进行入临界区。看看是什么实现，其实Semaphore的内部结构和ReentrantLock差不多，主要就是看看他是如何利用AbstractQueuedSynchronizer来实现并发的。首先看看如何利用tryAcquireShared()控制并发线程数的。

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

　　

是不是很简单，就是利用CAS来改变available值来控制，而available就是我们自己定义的能够进入临界区的最大线程数 那么releaseShared()是怎么实现的，其实简单一想也应该是增加available值

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))
            return true;
    }
}

　

## 小节
可以看出AbstractQueuedSynchronizer主要用分为共享锁和独占锁，对于共享锁就像是一个计数器，每次获取锁的时候计数器-1，释放的时候计数器+1。

 

ReentrantReadWriteLock

ReentrantReadWriteLock是读写锁，每个读锁之间是不竞争锁的，但读锁和写锁、写锁和写锁之间是竞争关系。这个并发类就是综合利用上面所说的内容。首先看一下内部结构

private final ReentrantReadWriteLock.ReadLock readerLock;

private final ReentrantReadWriteLock.WriteLock writerLock;

final Sync sync;

　　

成员变量分别是读锁和写锁（内部类）和一个继承了AbstractQueuedSynchronizer的Sync类，如果理解了上面的ReentrantLock和Semaphore,我们自己进行推敲一下，应该是读锁应该是和Semaphore类似，而写锁应该是和ReentrantLock类似。虽然这么说，还是有许多细节需要注意的：

• 如果保证非公平锁下，写锁不会饿死？
• 一个AbstractQueuedSynchronizer只有一个state字段，那么是如何保存读写的数量和写锁的数量呢？
• 对于读锁，如何知道当前自己持有多少个单位的锁呢

上面都是需要在设计读写锁需要考虑的地方，相信你看了源码就会知道答案！