Анализ исходного кода CountDownLatch серии Dead Java Synchronization

🖕Приглашаем обратить внимание на мой публичный аккаунт «Брат Тонг читает исходный код», ознакомиться с другими статьями из серии исходного кода и поплавать в океане исходного кода вместе с братом Тонгом.

(удобнее просматривать исходный код на горизонтальном экране мобильного телефона)

проблема

(1) Что такое CountDownLatch?

(2) Какие функции есть у CountDownLatch?

(3) В каких сценариях обычно используется CountDownLatch?

(4) Можно ли настроить начальный счет CountDownLatch?

Введение

CountDownLatch можно перевести как таймер обратного отсчета, но это кажется неточным, его смысл в том, чтобы позволить одному или нескольким потокам дождаться завершения операций других потоков перед выполнением последующих операций.

Обычное использование CountDownLatch несколько похоже на Thread.join(), ожидая завершения других потоков перед выполнением основной задачи.

структура класса

CountDownLatch содержит только внутренний класс Sync, у него нет справедливого/несправедливого режима, поэтому это относительно простой синхронизатор.

Также обратите внимание, что CountDownLatch не реализует интерфейс Serializable, поэтому его нельзя сериализовать.

Анализ исходного кода

Синхронизация внутреннего класса

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;
    
    // 传入初始次数
    Sync(int count) {
        setState(count);
    }
    // 获取还剩的次数
    int getCount() {
        return getState();
    }
    // 尝试获取共享锁
    protected int tryAcquireShared(int acquires) {
        // 注意，这里state等于0的时候返回的是1，也就是说count减为0的时候获取总是成功
        // state不等于0的时候返回的是-1，也就是count不为0的时候总是要排队
        return (getState() == 0) ? 1 : -1;
    }
    // 尝试释放锁
    protected boolean tryReleaseShared(int releases) {
        for (;;) {
            // state的值
            int c = getState();
            // 等于0了，则无法再释放了
            if (c == 0)
                return false;
            // 将count的值减1
            int nextc = c-1;
            // 原子更新state的值
            if (compareAndSetState(c, nextc))
                // 减为0的时候返回true，这时会唤醒后面排队的线程
                return nextc == 0;
        }
    }
}

Синхронизация переопределяет методы tryAcquireShared() и tryReleaseShared() и сохраняет количество в переменной состояния.

Здесь следует отметить, что параметры двух вышеуказанных методов бесполезны.

Метод строительства

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

Конструктор должен пройти счет, который является начальным числом.

метод ожидания()

// java.util.concurrent.CountDownLatch.await()
public void await() throws InterruptedException {
    // 调用AQS的acquireSharedInterruptibly()方法
    sync.acquireSharedInterruptibly(1);
}
// java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly()
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 尝试获取锁，如果失败则排队
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

Метод await() – это метод ожидания завершения других потоков. Сначала он попытается получить общую блокировку. В случае неудачи он войдет в очередь AQS и встанет в очередь для пробуждения.

Согласно приведенному выше исходному коду Sync, мы знаем, что когда состояние не равно 0, tryAcquireShared() возвращает -1, что означает, что все потоки, вызывающие метод await(), должны быть поставлены в очередь, когда счетчик не уменьшается до 0. .

Метод обратного отсчета()

// java.util.concurrent.CountDownLatch.countDown()
public void countDown() {
    // 调用AQS的释放共享锁方法
    sync.releaseShared(1);
}
// java.util.concurrent.locks.AbstractQueuedSynchronizer.releaseShared()
public final boolean releaseShared(int arg) {
    // 尝试释放共享锁，如果成功了，就唤醒排队的线程
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

Метод countDown() снимет общую блокировку, то есть количество счетчиков будет уменьшено на 1.

В соответствии с вышеупомянутым исходным кодом синхронизации, мы знаем, что TRERELEasshared () будет подсчитать количество раз каждого минус 1, возвращает True, когда его уменьшается до 0, на этот раз пробудут ожидающие потоки.

Обратите внимание, что doReleaseShared() пробуждает ожидающий поток, который мы проанализировали в предыдущей главе.

Случаи применения

Здесь мы имитируем сценарий использования, у нас есть основной поток и 5 вспомогательных потоков, ждем готовности основного потока, начинают выполняться 5 вспомогательных потоков, ждем завершения выполнения 5 вспомогательных потоков, основной поток продолжает выполняться, примерно блок-схема выглядит следующим образом:

Давайте посмотрим, как должен быть написан этот код:

public class CountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(5);

        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try {
                    System.out.println("Aid thread is waiting for starting.");
                    startSignal.await();
                    // do sth
                    System.out.println("Aid thread is doing something.");
                    doneSignal.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        // main thread do sth
        Thread.sleep(2000);
        System.out.println("Main thread is doing something.");
        startSignal.countDown();

        // main thread do sth else
        System.out.println("Main thread is waiting for aid threads finishing.");
        doneSignal.await();

        System.out.println("Main thread is doing something after all threads have finished.");

    }
}

Этот код разделен на две части:

Первый сегмент, сигнал 5 вспомогательных потоков, ожидающих запуска, сигнал, испускаемый основным потоком, рабочий поток вызывает 5 метод startSignal.await(), ожидающий сигнала запуска, главное, когда поток завершится, вызов startSignal.countDown () рабочий поток уведомления о начале работы.

Во втором абзаце основной поток ожидает сигнала завершения 5 вспомогательных потоков, а сигнал выдается 5 вспомогательными потоками, поэтому основной поток вызывает метод doneSignal.await() для ожидания сигнала завершения и вызывает, когда 5 вспомогательных потоков завершают свою работу Метод doneSignal.countDown() отправляет собственный сигнал завершения, когда сигнал завершения достигает 5, он пробуждает основной поток для продолжения выполнения последующей логики.

Суммировать

(1) CountDownLatch означает, что одному или нескольким потокам разрешено ждать завершения операций других потоков перед выполнением последующих операций;

(2) CountDownLatch реализован с использованием механизма общей блокировки AQS;

(3) Когда CountDownLatch инициализируется, необходимо передать количество отсчетов;

(4) Количество вызовов метода countDown() каждый раз уменьшается на 1;

(5) await () метод, когда каждый вызов попытается получить блокировку, замок фактически приобретается здесь, чтобы проверить значение AQS в переменной 1;

(6) Когда значение count (то есть значение состояния) уменьшится до 0, потоки в очереди будут пробуждены (эти потоки вызывают await() для входа в очередь);

пасхальные яйца

(1) Можно ли настроить начальный счет CountDownLatch?

A: Когда мы изучали Semaphore раньше, мы обнаружили, что время его лицензии можно изменить в любое время.Таким образом, можно ли изменить начальное время CountDownLatch в любое время? Ответ — нет, он не дает возможности изменять (увеличивать или уменьшать) количество раз, за исключением использования отражения для обмана.

(2) Почему CountDownLatch использует общие блокировки?

Ответ: мы изучили режим разделяемой блокировки AQS, когда ранее анализировали ReentrantReadWriteLock. Например, текущая блокировка приобретается потоком как мьютекс, тогда все потоки, которым необходимо получить общую блокировку, должны попасть в очередь AQS для постановки в очередь. this Когда мьютекс освобождается, он пробуждает эти непрерывные потоки в очереди, ожидающие получения общей блокировки один за другим.Обратите внимание, что термин здесь «пробуждается один за другим», что означает, что эти потоки, ожидающие получения общей блокировки не являются одноразовым пробуждением.

Кстати говоря, вы понимаете? Поскольку несколько потоков CountDownLatch’s await() могут вызываться несколько раз, при многократном вызове эти потоки должны входить в очередь AQS для постановки в очередь. .Если невозможно использовать мьютекс.Мьютексы являются взаимоисключающими среди нескольких потоков, и одновременно может быть разбужен только один.Нет гарантии, что, когда счетчик уменьшится до 0, все потоки, вызвавшие Метод ожидания() для ожидания будет разбужен.

(3) В чем разница между CountDownLatch и Thread.join()?

Ответ: Thread.join() вызывается в основном потоке. Он может только дождаться завершения вызываемого потока, прежде чем уведомлять основной поток, в то время как CountDownLatch отличается. Его метод countDown() может быть вызван в любой момент выполнения потока. ., больше гибкости.