兜里有辣条

兜里有辣条

关注
发信
关注(28)粉丝(399)

Java同步机制的底层实现

在多线程编程中我们会遇到很多需要使用线程同步机制去解决的并发问题，而这些同步机制就是多线程编程中影响正确性和运行效率的重中之重。这不禁让我感到好奇，这些同步机制是如何实现的呢？好奇心是进步的源泉，就让我们一起来揭开同步机制源码的神秘面纱吧。

在本文中，我们会从JDK中大多数同步机制的共同基础AbstractQueuedSynchronizer类开始说起，然后通过源码了解我们最常用的两个同步类可重入锁ReentrantLock和闭锁CountDownLatch的具体实现。通过这篇文章我们将可以了解到ReentrantLock和CountDownLatch两个常用同步类的源代码实现，并且掌握阅读其他基于AQS实现的同步工具类源码的能力，甚至可以利用AQS写出自己的同步工具类。

阅读这篇文章需要了解基本的线程同步机制，有兴趣的读者可以参考一下这篇文章《多线程中那些看不到的陷阱》。

同步机制的核心——AQS

同步机制源码初探

ReentrantLock是我们常用的一种可重入互斥锁，是synchronized关键字的一个很好的替代品。互斥指的就是同一时间只能有一个线程获取到这个锁，而可重入是指如果一个线程再次获取一个它已经持有的互斥锁，那么仍然会成功。

这个类的源码在JDK的java.util.concurrent包下，我们可以在IDE中点击类名跳转到具体的类定义，比如下面就是在我的电脑上跳转之后看到的ReentrantLock类的源代码。在这里我们可以看到在ReentrantLock类中还包含了一个继承自AbstractQueuedSynchronizer类的内部类，而且有一个该内部类Sync类型的字段sync。实际上ReentrantLock类就是通过这个内部类对象来实现线程同步的。

如果打开CountDownLatch的源代码，我们会发现这个类里也同样有一个继承自AbstractQueuedSynchronizer类的子类Sync，并且也有一个Sync类型的字段sync。在java.util.concurrent包下的大多数同步工具类的底层都是通过在内部定义一个AbstractQueuedSynchronizer类的子类来实现的，包括我们在本文中没提到的许多其他常用类也是如此，比如：读写锁ReentrantReadWriteLock、信号量Semaphore等。

AQS是什么？

那么这个AbstractQueuedSynchronizer类也就是我们所说的AQS，到底是何方神圣呢？这个类首先像我们上面提到的，是大多数多线程同步工具类的基础。它内部包含了一个对同步器的等待队列，其中包含了所有在等待获取同步器的线程，在这个等待队列中的线程将会在同步器释放时被唤醒。比如一个线程在获取互斥锁失败时就会被放入到等待队列中等待被唤醒，这也就是AQS中的Q——“Queued”的由来。

而类名中的第一个单词Abstract是因为AQS是一个抽象类，它的使用方法就是实现继承它的子类，然后使用这个子类类型的对象。在这个子类中我们会通过重写下列的五个方法中的一部分或者全部来指定这个同步器的行为策略：

  1. boolean tryAcquire(int arg)，独占式获取同步器，独占式指同一时间只能有一个线程获取到同步器；
  2. boolean tryRelease(int arg)，独占式释放同步器；
  3. boolean isHeldExclusively()，同步器是否被当前线程独占式地持有；
  4. int tryAcquireShared(int arg)，共享式获取同步器，共享式指的是同一时间可能有多个线程同时获取到同步器，但是可能会有数量的限制；
  5. boolean tryReleaseShared(int arg)，共享式释放同步器。

这五个方法之所以能指定同步器的行为，则是因为AQS中的其他方法就是通过对这五个方法的调用来实现的。比如在下面的acquire方法中就调用了tryAcquire来获取同步器，并且在被调用的acquireQueued方法内部也是通过tryAcquire方法来循环尝试获取同步器的。

public final void acquire(int arg) {
    // 1. 调用tryAcquire方法尝试获取锁
    // 2. 如果获取失败(tryAcquire返回false)，则调用addWaiter方法将当前线程保存到等待队列中
    // 3. 之后调用acquireQueued方法来循环执行“获取同步器 -> 获取失败休眠 -> 被唤醒重新获取”过程
    //      直到成功获取到同步器返回false；或者被中断返回true
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 如果acquireQueued方法返回true说明线程被中断了
        //   所以调用selfInterrupt方法中断当前线程
        selfInterrupt();
}

下面，我们就来看看在ReentrantLock和CountDownLatch两个类中定义的AQS子类到底是如何重写这五个方法的。

CountDownLatch的实现

CountDownLatch是一种典型的闭锁，比如我需要使用四个线程完成四种不同的计算，然后把四个线程的计算结果相加后返回，这种情况下主线程就需要等待四个完成不同任务的工作线程完成之后才能继续执行。那么我们就可以创建一个初始的count值为4的CountDownLatch，然后在每个工作线程完成任务时都对这个CountDownLatch执行一个countDown操作，这样CountDownLatch中的count值就会减1。当count值减到0时，主线程就会从阻塞中恢复，然后将四个任务的结果相加后返回。

下面是CountDownLath的几个常用方法：

  1. void await()，等待操作，如果count值目前已经是0了，那么就直接返回；否则就进入阻塞状态，等待count值变为0；
  2. void countDown()，减少计数操作，会让count减1。

调用多次countDown()方法让count值变为0之后，被await()方法阻塞的线程就可以继续执行了。了解了CountDownLatch的基本用法之后我们就来看看这个闭锁到底是怎么实现的，首先，我们来看一下CountDownLatch中AQS的子类，内部类Sync的定义。

CountDownLatch的内部Sync类

下面的代码是CountDownLatch中AQS的子类Sync的定义，Sync是CountDownLatch类中的一个内部类。在这个类中重写了AQS的tryAcquireShared和tryReleaseShared两个方法，这两个都是共享模式需要重写的方法，因为CountDownLatch在count值为0时可以被任意多个线程同时获取成功，所以应该实现共享模式的方法。

在CountDownLatch的Sync中使用了AQS的state值用来存放count值，在初始化时会把state值初始化为n。然后在调用tryReleaseShared时会将count值减1，但是因为这个方法可能会被多个线程同时调用，所以要用CAS操作保证更新操作的原子性，就像我们用AtomicInteger一样。在CAS失败时我们需要通过重试来保证把state减1，如果CAS成功时，即使有许多线程同时执行这个操作最后的结果也一定是正确的。在这里，tryReleaseShared方法的返回值表示这个释放操作是否可以让等待中的线程成功获取同步器，所以只有在count为0时才能返回true。

tryAcquireShared方法就比较简单了，直接返回state是否等于0即可，因为只有在CountDownLatch中的count值为0时所有希望获取同步器的线程才能获取成功并继续执行。如果count不为0，那么线程就需要进入阻塞状态，等到count值变为0才能继续执行。

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

    // 构造器，初始化count值
    // 在这个子类中把count值保存到了AQS的state中
    Sync(int count) {
        setState(count);
    }

    // 获取当前的count值
    int getCount() {
        return getState();
    }

    // 获取操作在state为0时会成功，否则失败
    // tryAcquireShared失败时，线程会进入阻塞状态等待获取成功
    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

    // 对闭锁执行释放操作减小计数值
    protected boolean tryReleaseShared(int releases) {
        // 减小coun值，在count值归零时唤醒等待的线程
        for (;;) {
            int c = getState();

            // 如果计数已经归零，则直接释放失败
            if (c == 0)
                return false;

            // 将计数值减1
            int nextc = c-1;
            // 为了线程安全，以CAS循环尝试更新
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}

CounDownLatch对Sync类对象的使用

看了CountDownLatch中的Sync内部类定义之后，我们再来看看CountDownLatch是如何使用这个内部类的。

在CountDownLatch的构造器中，初始化CountDownLatch对象时会同时在其内部初始化保存一个Sync类型的对象到sync字段用于之后的同步操作。并且传入Sync类构造器的count一定会大于等于0。

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

有了Sync类型的对象之后，我们在await()方法里就可以直接调用sync的acquireSharedInterruptibly方法来获取同步器并陷入阻塞，等待count值变为0了。在AQS的acquireSharedInterruptibly方法中会在调用我们重写的tryAcquireShared方法获取失败时进入阻塞状态，直到CountDownLatch的count值变为0时才能成功获取到同步器。

public void await() throws InterruptedException {
    // 调用sync对象的获取方法来进入锁等待
    sync.acquireSharedInterruptibly(1);
}

而在CountDownLatch的另一个减少count值的重要方法countDown()中，我们同样是通过调用sync上的方法来实现具体的同步功能。在这里，AQS的releaseShared(1)方法中同样会调用我们在Sync类中重写的tryReleaseShared方法来执行释放操作，并在tryReleaseShared方法返回true时去唤醒等待队列中的阻塞等待线程，让它们在count值为0时能够继续执行。

public void countDown() {
    sync.releaseShared(1);
}

从上文中可以看出，CoundDownLatch中的各种功能都是通过内部类Sync来实现的，而这个Sync类就是一个继承自AQS的子类。通过在内部类Sync中重写了AQS的tryAcquireShared和tryReleaseShared两个方法，我们就指定了AQS的行为策略，使其能够符合我们对CountDownLatch功能的期望。这就是AQS的使用方法，下面我们来看一个大家可能会更熟悉的例子，来进一步了解AQS在独占模式下的用法。

ReentrantLock的实现

可重入锁ReentrantLock可以说是我们的老朋友了，从最早的synchronized关键字开始，我们就开始使用类似的功能了。可重入锁的特点主要有两点：

  1. 同一时间只能有一个线程持有

    • 如果我想保护一段代码同一时间只能被一个线程所访问，比如对一个队列的插入操作。那么如果有一个线程已经获取了锁之后在修改队列了，那么其他也想要修改队列的线程就会陷入阻塞，等待之前的这个线程执行完成。

  2. 同一线程可以对一个锁重复获取成功多次

    • 而如果一个线程对同一个队列执行了两个插入操作，那么第二次获取锁时仍然会成功，而不会被第一次成功获取到的锁所阻塞。

ReentrantLock类的常用操作主要有三种：

  1. 获取锁，一个线程一旦获取锁成功后就会阻塞其他线程获取同一个锁的操作，所以一旦获取失败，那么当前线程就会被阻塞

    • 最简单的获取锁方法就是调用public void lock()方法

  2. 释放锁，获取锁之后就要在使用完之后释放它，否则别的线程都将会因无法获取锁而被阻塞，所以我们一般会在finally中进行锁的释放操作

    • 可以通过调用ReentrantLock对象的unlock方法来释放锁

  3. 获取条件变量，条件变量是和互斥锁搭配使用的一种非常有用的数据结构，有兴趣的读者可以通过《从0到1实现自己的阻塞队列(上)》这篇文章来了解条件变量具体的使用方法

    • 我们可以通过Condition newCondition()方法来获取条件变量对象，然后调用条件变量对象上的await()、signal()、signalAll()方法来进行使用

ReentrantLock的内部Sync类

在ReentrantLock类中存在两种AQS的子类，一个实现了非公平锁，一个实现了公平锁。所谓的“公平”指的就是获取互斥锁成功返回的时间会和获取锁操作发起的时间顺序一致，例如有线程A已经持有了互斥锁，当线程B、C、D按字母顺序获取锁并进入等待，线程A释放锁后一定是线程B被唤醒，线程B释放锁后一定是C先被唤醒。也就是说锁被释放后对等待线程的唤醒顺序和获取锁操作的顺序一致。而且如果在这个过程中，有其他线程发起了获取锁操作，因为等待队列中已经有线程在等待了，那么这个线程一定要排到等待队列最后去，而不能直接抢占刚刚被释放还未被刚刚被唤醒的线程锁持有的锁。

下面我们同样先看一下ReentrantLock类中定义的AQS子类Sync的具体源代码。下面是上一段说到的非公平Sync类和公平Sync类两个类的共同父类Sync的带注释源代码，里面包含了大部分核心功能的实现。虽然下面包含了该类完整的源代码，但是我们现在只需要关心三个核心操作，也是我们在独占模式下需要重写的三个AQS方法：tryAcquire、tryRelease和isHeldExclusively。建议在看完文章之后再回来回顾该类中其他的方法实现，直接跳过其他的方法当然也是完全没有问题的。

abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = -5179523762034025860L;

    /**
     * 实现Lock接口的lock方法，子类化的主要原因是为了非公平版本的快速实现
     */
    abstract void lock();

    /**
     * 执行非公平的tryLock。tryAcquire方法在子类中被实现，但是两者都需要非公平版本的trylock方法实现。
     */
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();

        // 如果锁还未被持有
        if (c == 0) {
            // 通过CAS尝试获取锁
            if (compareAndSetState(0, acquires)) {
                // 如果锁获取成功则将锁持有者改为当前线程，并返回true
                setExclusiveOwnerThread(current);
                return true;
            }
        }

        // 锁已经被持有，则判断锁的持有者是否是当前线程
        else if (current == getExclusiveOwnerThread()) {
            // 可重入锁，如果锁的持有者是当前线程，那就在state上加上新的获取数
            int nextc = c + acquires;

            // 判断新的state值有没有溢出
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");

            // 将新的state更新为新的值，因为可以进入这段代码的只有一个线程
            // 所以不需要线程安全措施
            setState(nextc);
            return true;
        }

        return false;
    }

    // 重写了AQS的独占式释放锁方法
    protected final boolean tryRelease(int releases) {
        // 计算剩余的锁持有量
        // 因为只有当前线程持有该锁的情况下才能执行这个方法，所以不需要做多线程保护
        int c = getState() - releases;

        // 如果当前线程未持有锁，则直接抛出错误
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();

        boolean free = false;
        // 如果锁持有数已经减少到0，则释放该锁，并清空锁持有者
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }

        // 更新state值，只有state值被设置为0才是真正地释放了锁
        // 所以setState和setExclusiveOwnerThread之间不需要额外的同步措施
        setState(c);
        return free;
    }

    // 当前线程是否持有该锁
    protected final boolean isHeldExclusively() {
        return getExclusiveOwnerThread() == Thread.currentThread();
    }

    // 创建对应的条件变量
    final ConditionObject newCondition() {
        return new ConditionObject();
    }

    // 从外层传递进来的方法

    // 获取当前的锁持有者
    final Thread getOwner() {
        return getState() == 0 ? null : getExclusiveOwnerThread();
    }

    // 获取锁的持有计数
    // 如果当前线程持有了该锁则返回state值，否则返回0
    final int getHoldCount() {
        return isHeldExclusively() ? getState() : 0;
    }

    // 判断锁是否已经被持有
    final boolean isLocked() {
        return getState() != 0;
    }

}

实际的tryAcquire方法将在公平Sync类与非公平Sync类两个子类中实现，但是这两个子类都需要调用父类Sync中的非公平版本的tryAcquire——nonfairTryAcquire方法。在这个方法中，我们主要做两件事：

  1. 当前锁还未被人持有。在ReentrantLock中使用AQS的state来保存锁的状态，state等于0时代表锁没有被任何线程持有，如果state大于0，那么就代表持有者对该锁的重复获取次数

    • 如果当前锁还未被线程持有，那么就会通过compareAndSetState来原子性地修改state值，修改成功则需要设置当前线程为锁的持有线程并返回true代表获取成功；否则就返回

  2. 锁已被当前线程持有

    • 在锁已被当前线程持有的情况下，就需要将state值加1代表持有者线程对锁的重复获取次数。

而对于独占式释放同步器的tryRelease方法，则在父类Sync中直接实现了，两个公平/非公平子类调用的都是同一段代码。首先，只有锁的持有者才能释放锁，所以如果当前线程不是所有者线程在释放操作中就会抛出异常。如果释放操作会将持有计数清零，那么当前线程就不再是该锁的持有者了，锁会被完全释放，而锁的所有者会被设置为null。最后，Sync会将减掉入参中的释放数之后的新持有计数更新到AQS的state中，并返回锁是否已经被完全释放了。

isHeldExclusively方法比较简单，它只是检查锁的持有者是否是当前线程。

非公平Sync类的实现

Sync的两个公平/非公平子类的实现比较简单，下面是非公平版本子类的源代码。在非公平版本的实现中，调用lock方法首先会尝试通过CAS修改AQS的state值来直接抢占锁，如果抢占成功就直接将持有者设置为当前线程；如果抢占失败就调用acquire方法走正常流程来获取锁。而在acquire方法中就会调用子类中的tryAcquire方法并进一步调用到上文提到的父类中的nonfairTryAcquire方法来完成锁获取操作。

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * 执行锁操作。尝试直接抢占，如果失败的话就回到正常的获取流程进行
     */
    final void lock() {
        // 尝试直接抢占
        if (compareAndSetState(0, 1))
            // 抢占成功设置锁所有者
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 抢占失败走正常获取流程
            acquire(1);
    }

    // 实现AQS方法，使用nonfairTryAcquire实现
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

公平Sync类的实现

而在公平版本的Sync子类FairSync中，为了保证成功获取到锁的顺序一定要和发起获取锁操作的顺序一致，所以自然不能在lock方法中进行CAS方式的抢占，只能老老实实调用acquire方法走正式流程。而acquire方法最终就会调用子类中定义的tryAcquire来真正获取锁。

在tryAcquire方法中，代码主要处理了两种情况：

  1. 当前锁还没有被线程锁持有

    • 只有在确保等待队列为空的情况下才能尝试用CAS方式直接抢占锁，而在等待队列不为空的情况下，最后返回了false，之后acquire方法中的代码会将当前线程放入到等待队列中阻塞等待锁的释放。这就保证了在获取锁时已经有线程等待的情况下，任何线程都要进入等待队列去等待获取锁，而不能直接对锁进行获取。

  2. 当前线程已经持有了该锁

    • 如果当前线程已经是该锁的持有者了，那么就会在state值上加上本次的获取数量来更新锁的重复获取次数，并返回true代表获取锁成功。
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    // 直接使用acquire进行获取锁操作
    final void lock() {
        acquire(1);
    }

    /**
     * 公平版本的tryAcquire方法。不要授予访问权限，除非是递归调用或者没有等待线程或者这是第一个调用
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();

        // 如果锁没有被持有
        if (c == 0) {
            // 为了实现公平特性，所以只有在等待队列为空的情况下才能直接抢占
            // 否则只能进入队列等待
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }

        // 如果锁已被持有，且当前线程就是持有线程
        else if (current == getExclusiveOwnerThread()) {
            // 计算新的state值
            int nextc = c + acquires;

            // 如果锁计数溢出，则抛出异常
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");

            // 设置state状态值
            setState(nextc);
            return true;
        }
        return false;
    }
}

ReentrantLock对Sync类对象的使用

最后，我们来看看ReentrantLock类中的lock()、unlock()、newCondition方法对Sync类对象的使用方式。

首先是在构造器中，根据入参指定的公平/非公平模式创建不同的内部Sync类对象，如果是公平模式就是用FairSync类，如果是非公平模式就是用NonfairSync类。

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

然后在互斥锁的锁定方法lock()中，ReentrantLock直接使用Sync类中的lock方法来实现了锁的获取功能。

public void lock() {
    // 调用sync对象的lock方法实现
    sync.lock();
}

在unlock()方法中也是一样的情况，ReentrantLock直接依赖Sync类对象来实现这个功能。

public void unlock() {
    // 调用了sync对象的release方法实现
    sync.release(1);
}

最后一个创建条件变量的方法则直接依赖于AQS中定义的方法，我们在ReentranctLock的Sync类中并不需要做任务额外的工作，AQS就能为我们做好所有的事情。

public Condition newCondition() {
    // 调用了sync对象继承自AQS的`newCondition`方法实现
    return sync.newCondition();
}

通过ReentrantLock的例子我们能够更明显地感受到，这些基于AQS实现同步功能的类中并不需要做太多额外的工作，大多数操作都是通过直接调用Sync类对象上的方法来实现的。只要定义好了继承自AQS的子类Sync，并通过Sync类重写几个AQS的关键方法来指定AQS的行为策略，就可以实现风格迥异的各种同步工具类了。

总结

在这篇文章中，我们从AQS的基本概念说起，简单介绍了AQS的具体用法，然后通过CountDownLatch和ReentrantLock两个常用的多线程同步工具类的源码来具体了解了AQS的使用方式。我们不仅可以完全弄明白这两个线程同步类的实现原理与细节，而且最重要的是找到了AQS这个幕后大BOSS。通过AQS，我们不仅可以更容易地阅读并理解其他同步工具类的使用与实现，而且甚至可以动手开发出我们自己的自定义同步工具类。

到了这里，这一系列多线程编程相关的技术文章就接近尾声了。后续我还会发布一篇囊括这个系列所有内容的总结性文章，里面会对多线程编程相关的知识脉络做一次全面的梳理，然后将每个知识点链接到具体阐释这个主题的文章中去。让读者可以在宏观和微观两个层面理解多线程编程的原理与技巧，帮助大家建立完整的Java多线程理论与实践知识体系。有兴趣的读者可以关注一下后续的文章，感谢大家的支持。

05-26 19:34
Powered by LMLPHP ©2024 兜里有辣条 0.040383