Подробное объяснение исходного кода AQS

Java
Подробное объяснение исходного кода AQS

Что такое АКС?

AQS(AbstractQueuedSynchronizer) Основная идея:Если запрошенный общий ресурс свободен, поток, в данный момент запрашивающий ресурс, устанавливается как допустимый рабочий поток, а общий ресурс устанавливается в заблокированном состоянии. Если запрошенные разделяемые ресурсы заняты, то требуется набор механизмов блокировки потоков в ожидании и выделения блокировок при пробуждении.Этот механизм AQS реализуется блокировками очереди CLH, то есть в очередь добавляются потоки, которые не могут временно получить блокировки.

Очередь CLH (Craig, Landin, and Hagersten) — это виртуальная двусторонняя очередь (виртуальная двусторонняя очередь означает отсутствие экземпляра очереди, а только связь между узлами). AQS инкапсулирует каждый поток, запрашивающий общие ресурсы, в узел Node очереди блокировок CLH для обеспечения выделения блокировок.

Очередь CLH показана на рисунке:

синхронизатор очередиAbstractQueuedSynchronizer(далее именуемый синхронизатором) — это базовая структура, используемая для создания блокировок или других компонентов синхронизации, она используетintпеременная-член типаstateУказывает состояние синхронизации, и работа по организации очереди потока получения ресурсов завершена через встроенную очередь FIFO.

Основное использование синхронизатора - наследование.Подкласс управляет состоянием синхронизации, наследуя синхронизатор и реализуя его абстрактный метод.Во время реализации абстрактного метода неизбежно изменение состояния синхронизации.В этом случае нужно использовать состояние синхронизации, предоставляемое синхронизатором. 3 методаgetState(),setState(int newState)иcompareAndSetState(int expect, int update)для выполнения операций, потому что они гарантируют, что изменения состояния безопасны.

Рекомендация подкласса определяется как настраиваемый компонент синхронизации.статический внутренний класс, сам синхронизатор не реализует какой-либо интерфейс синхронизации, он просто определяет ряд методов получения и освобождения состояния синхронизации для использования пользовательских компонентов синхронизации, синхронизатор может поддерживать обаэксклюзивный доступ к статусу синхронизации, также может поддерживатьОбщий доступ к статусу синхронизации, чтобы было удобно реализовывать разные типы компонентов синхронизации (ReentrantLock,ReentrantReadWriteLockиCountDownLatchЖдать).

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

Конструкция синхронизатора основана наШаблон метода шаблона, это,Пользователю необходимо наследовать синхронизатор и переопределить указанный метод, затем объединить синхронизатор в реализации пользовательского компонента синхронизации и вызвать методы шаблона, предоставляемые синхронизатором, и эти методы шаблона будут вызывать методы, переопределенные пользователем.

При переопределении метода, указанного синхронизатором, необходимо использовать следующие три метода, предоставляемые синхронизатором, для доступа или изменения состояния синхронизации:

  • getState(): получить текущий статус синхронизации
  • setState(int newState): установить текущее состояние синхронизации
  • compareAndSetState(int expect,int update): Используйте CAS для установки текущего состояния, этот метод может обеспечить атомарность настроек состояния.

Переопределяемые методы синхронизатора показаны в следующей таблице.

имя метода описывать
protected boolean tryAcquire(int arg) Получить исключительно статус синхронизации.Для реализации этого метода необходимо запросить текущий статус и определить, соответствует ли статус синхронизации ожиданиям, а затем выполнить CAS для установки статуса синхронизации.
protected boolean tryRelease(int arg) Состояние синхронизации высвобождается исключительно, и поток, ожидающий получения состояния синхронизации, будет иметь возможность получить состояние синхронизации.
protected int tryAcquireShared(int arg) Общий статус синхронизации сбора данных, возвращает значение, большее или равное 0, что указывает на то, что сбор данных выполнен успешно, в противном случае получение данных завершается неудачно.
protected boolean tryReleaseShared(int arg) Общее состояние синхронизации выпуска
protected boolean isHeldExclusively() Занят ли текущий синхронизатор потоком в эксклюзивном режиме, обычно этот метод указывает, занят ли он текущим потоком.

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

image-20200606212156471

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

Реализация эксклюзивной блокировки Mutex

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

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

MutexРеализация кода выглядит следующим образом:

public class Mutex implements Lock {
    // 静态内部类，自定义同步器
    private static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -4387327721959839431L;

        // 是否处于占用状态
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        // 当状态为0的时候获取锁
        @Override
        public boolean tryAcquire(int acquires) {
            assert acquires == 1; // Otherwise unused
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        // 释放锁，将状态设置为0
        @Override
        protected boolean tryRelease(int releases) {
            assert releases == 1; // Otherwise unused
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        // 返回一个Condition，每个condition都包含了一个condition队列
        Condition newCondition() {
            return new ConditionObject();
        }
    }

    // 仅需要将操作代理到Sync上即可
    private final Sync sync = new Sync();

    @Override
    public void lock() {
        sync.acquire(1);
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    public Condition newCondition() {
        return sync.newCondition();
    }

    public boolean isLocked() {
        return sync.isHeldExclusively();
    }

    public boolean hasQueuedThreads() {
        return sync.hasQueuedThreads();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
}

В приведенном выше примере эксклюзивная блокировкаMutexЯвляется настраиваемым компонентом синхронизации, который позволяет одновременно удерживать блокировку только одному потоку.MutexОпределяет статический класс внутреннего класса, который наследует синхронизатор и реализует эксклюзивное приобретение и освобождение состояния синхронизации. существуетtryAcquire(int acquires)В методе, если настройка CAS прошла успешно (статус синхронизации установлен в 1), это означает, что статус синхронизации получен, а вtryRelease(int releases)метод просто сбрасывает состояние синхронизации на 0. использование пользователемMutexне имеет прямого отношения к реализации внутреннего синхронизатора, а вместо этого вызываетMutexпредоставленный метод, вMutexреализация для получения блокировкиlock()метод в качестве примера, вам нужно только вызвать шаблонный метод синхронизатора в реализации методаacquire(int args)То есть, если текущему потоку не удастся вызвать этот метод для получения состояния синхронизации, он будет добавлен в очередь синхронизации и ожидать, что значительно снижает порог для реализации надежного пользовательского компонента синхронизации.

Принцип AQS

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

очередь синхронизации

Синхронизатор полагается на внутреннюю очередь синхронизации (двусторонняя очередь FIFO) для завершения управления состоянием синхронизации.Когда текущему потоку не удается получить состояние синхронизации, синхронизатор создает текущий поток, состояние ожидания и другую информацию. вNodeузла и добавить его в очередь синхронизации, при этом будетзаблокировать текущий поток, когда состояние синхронизации выделяется,Поток пробуждается, чтобы снова попытаться получить статус синхронизации.

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

NodeУ узла есть следующие состояния:

// 因为超时或者中断，节点会被设置成取消状态，被取消的节点不会参与到竞争中，保持取消状态不会改变
static final int CANCELLED =  1;

// 后继节点的线程处于等待状态，而当前节点的线程如果释放了同步状态或者被取消了，将会通知后继节点，使后继节点的线程得以运行
static final int SIGNAL    = -1;

// 表示节点在等待队列中，节点线程等待在Condition上，当其他线程对Condition调用了signal方法后，该节点将会从等待队列中转移到同步队列中，加入到对同步状态的获取中
static final int CONDITION = -2;

// 表示下一次共享式同步状态获取将会无条件被传播下去
static final int PROPAGATE = -3;

NodeСуществуют также различные узлы, определенные в:

// 后继节点
volatile Node next;

// 前驱结点，当节点加入同步队列时被设置（尾部添加）
volatile Node prev;

// 获取同步状态的线程
volatile Thread thread;

// 等待队列中的后继节点。如果当前节点是共享的，那么该字段是一个SHARED常量，也就是说节点类型（独占和共享）和等待队列中的后继节点共有一个字段
Node nextWaiter;

Узел является основой составляющих синхронных очередей, а синхронизатор имеет головные узлы.headи хвостовой узелtail, поток, которому не удалось получить состояние синхронизации, станет хвостом узла, присоединившегося к очереди.Основная структура очереди синхронизации показана на следующем рисунке.

На приведенной выше диаграмме синхронизатор содержит две ссылки типа узла, одну на головной узел, а другую на хвостовой узел. Представьте себе, что когда поток успешно получает состояние синхронизации, другие потоки не смогут получить состояние синхронизации, а вместо этого строятся как узлы и добавляются в очередь синхронизации, а процесс присоединения к очереди должен обеспечивать безопасность потоков, поэтому синхронизатор обеспечиваетУстановите хвостовой узел на основе CASМетоды:compareAndSetTail(Node expect, Node update), он должен передать хвостовой узел и текущий узел, который «думает» текущий поток.Только после успешной настройки текущий узел официально связывается с предыдущим хвостовым узлом.

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

В приведенной выше показателе узел Установленная головка завершена путем получения потока, успешно в синхронном состоянии. Поскольку только один поток может успешно приобретать синхронное состояние, метод настройки узла заголовка не должен использовать CAS, чтобы обеспечить его только только Нужно включить заголовок узел, чтобы стать последующим узлом и отключить исходный узел.nextПросто процитируйте.

Получение и освобождение эксклюзивного состояния синхронизации

вызовом синхронизатораacquire(int arg)Метод может получить состояние синхронизации.Этот метод не чувствителен к прерыванию, то есть, поскольку поток входит в очередь синхронизации после неудачной попытки получить состояние синхронизации, когда поток прерывается позже, поток не будет удален из синхронизации очередь.acquireКод метода следующий:

public final void acquire(int arg) {
    // 尝试获取锁
    if (!tryAcquire(arg) &&
        // 获取不到,则进入等待队列，返回是否中断
        // acquireQueued返回true表示中断
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 如果返回中断，则调用当前线程的interrupt()方法
        selfInterrupt();
}

Вышеприведенный код в основном выполняет связанную работу по получению состояния синхронизации, построению узла, присоединению к очереди синхронизации, а также вращению и ожиданию в очереди синхронизации.tryAcquire(int arg)Метод (метод, реализованный подклассом), потокобезопасный метод получения состояния синхронизации,Если не удается получить статус синхронизации, затем создайте узел синхронизации (эксклюзивныйNode.EXCLUSIVE, только один поток может одновременно успешно получить состояние синхронизации) и пройтиaddWaiter(Node node)Метод добавляет узел в хвост очереди синхронизации и, наконец, вызываетacquireQueued(Node node, int arg)метод, так что узел приобретает состояние синхронизации в режиме вращения. Звоните если нет в наличииLockSupport.park(this)Нить в блокирующем узле, и пробуждение заблокированного потока в основном зависит отУдалить узел-предшественник из очередиилиЗаблокированный поток прерываетсяреализовать.

Разберем сопутствующую работу, первая — построение узлов и постановка в очередь на синхронизацию.

private Node addWaiter(Node mode) {
    // 将当前线程封装成Node，并且mode为独占锁
    Node node = new Node(Thread.currentThread(), mode);
    // tail是AQS的中表示同步队列的队尾，刚开始为null，所以进行enq(node)方法
    Node pred = tail;
    if (pred != null) {
        // 将当前节点node的前驱结点设置为尾结点
        node.prev = pred;
        // CAS设置尾结点
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 第一次添加，tail=null，将node添加到同步队列中
    enq(node);
    return node;
}

private Node enq(final Node node) {
    // lockfree：循环+CAS
    for (;;) {
        Node t = tail;
        // 如果是第一次添加到队列，那么tail=null
        if (t == null) { 
            // CAS设置头节点
            if (compareAndSetHead(new Node()))
                tail = head;
        } 
        // 否则添加逻辑和addWaiter中相似
        // 既然相似，这里为什么需要再次判断不为null呢？
        // 假设有两个线程同时进入enq方法，先进来的线程判断 t==null 表示队列是首次使用，需要先初始化，
        // 那么第二个线程判断 if(t==null) 不通过，如果没有else逻辑，那么第二个线程就无法执行，
        // 因此需要和addWaiter相似的逻辑，保证多线程情况下也正常。
        else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

Приведенный выше код с использованиемcompareAndSetTail(Node expect, Node update)чтобы гарантировать, что узлы могут быть добавлены потокобезопасно.

существуетenq(final Node node)В методе синхронизатор обеспечивает правильное добавление узлов через «мертвую петлю», в «мертвой петле» текущий поток может вернуться из этого метода только после того, как узел будет установлен в качестве хвостового узла через CAS, в противном случае текущий поток продолжает пытаться установить . Как можно видеть,enq(final Node node)Метод «сериализует» параллельные запросы на добавление узлов через CAS.

После того, как узел попадает в очередь синхронизации, он входит в процесс вращения.Каждый узел (или каждый поток) наблюдает интроспективно.При выполнении условий и достижении состояния синхронизации он может выйти из этого процесса вращения.В противном случае он остается в процессе вращения. spin (и блокирует поток узла), а синхронизаторacquireQueuedМетод показан ниже.

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 其前驱是头结点，并且再次调用tryAcquire成功获取锁
            if (p == head && tryAcquire(arg)) {
                // 将自己设为头结点
                setHead(node);
                p.next = null; // help GC
                failed = false;
                // 成功获取锁，返回
                return interrupted;
            }
            // 没有得到锁时
            // shouldParkAfterFailedAcquire方法：返回是否需要阻塞当前线程
            // parkAndCheckInterrupt方法：阻塞当前线程，当线程再次唤醒时，返回是否被中断
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                // 修改中断标志位
                interrupted = true;
        }
    } finally {
        if (failed)
            // 获取锁失败，则将此线程对应的node的waitStatus改为CANCEL
            cancelAcquire(node);
    }
}

/*
 * 获取锁失败时，检查并更新node的waitStatus。
 * 如果线程需要阻塞，返回true。
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;

    // 前驱节点的waitStatus是SIGNAL
    if (ws == Node.SIGNAL)
        /* 
         * SIGNAL状态的节点，释放锁后，会唤醒其后继节点。
         * 因此，此线程可以安全的阻塞（前驱节点释放锁时，会唤醒此线程)。
         */
        return true;

    // 前驱节点对应的线程被取消，CANCELLED = 1
    if (ws > 0) {
        // 需要将取消状态的节点从队列中移除直到找到一个状态不是取消的节点为止
        do {        
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 除了以上情况，通过CAS将前驱节点的状态设置成SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

// 当shouldParkAfterFailedAcquire方法返回true，则调用parkAndCheckInterrupt方法阻塞当前线程
private final boolean parkAndCheckInterrupt() {
    // 阻塞当前线程
    LockSupport.park(this);
    // 判断当前线程是否被中断，如果中断了，则返回true
    // 由于Thread.interrupted()方法会清除中断标志位，所以后续需要修改还原成true
    return Thread.interrupted();
}

существуетacquireQueued(final Node node, int arg)В методе текущий поток пытается получить состояние синхронизации в «бесконечном цикле», и только узел-предшественник является головным узлом, который может попытаться получить состояние синхронизации по следующим двум причинам:

  • Головной узел — это узел, который успешно получил состояние синхронизации. После того, как поток головного узла сбросит состояние синхронизации, он разбудит свой узел-преемник. узел-предшественник является головным узлом.
  • Принцип FIFO для поддержания синхронных очередей

В этом методе поведение узла вращения для получения состояния синхронизации показано на следующем рисунке.

На приведенном выше рисунке, поскольку узел-предшественник потока неголовного узла удален из очереди или прерван, он возвращается из состояния ожидания, а затем проверяет, является ли его предшественник головным узлом, и если да, пытается получить состояние синхронизации. Видно, что узел и узел в основном не взаимодействуют друг с другом в процессе циклической проверки, а просто определяют, является ли их предшественник головным узлом, что приводит правило освобождения узла в соответствие с FIFO, а также облегчает преждевременное Обработка (преждевременное уведомление — это когда поток, узел-предшественник которого не является головным узлом, пробуждается из-за прерывания).

Эксклюзивный процесс получения состояния синхронизации:

Для каждого потока, который вы не можете получить блокировку, вы упаковываете себя.NodeУзел, присоединитесь к хвосту очереди, а затем оцените, является ли узел-предшественник текущего узла головным узлом, если да, попытайтесь получить блокировку, и если получение пройдет успешно, установите себя в качестве головного узла; если получение не удалось, продолжить проверку состояния узла-предшественника, если состояние дляSIGNAL(-1), текущий узел переходит в состояние блокировки (Узлы-предшественники могут просыпаться, уменьшая накладные расходы на вращение.); если кромеCANCELLED(1) кроме состояния, измените его наSIGNAL(-1). Когда добавляется новый узел, сначала прокрутите, чтобы определить, является ли узел-предшественник текущего узла головным узлом, если нет, непосредственно определите, является ли состояниеSIGNAL(-1), заблокировать, если он есть, и заблокировать, если он является дополнением кCANCELLED(1) кроме состояния, измените его наSIGNAL(-1).

Таким образом, весь процесс иногда вращается, иногда блокируется, пока блокировка не будет получена или отменена.

Соответствующие схемы см. в этой статье https://www.jianshu.com/p/b6efbdbdc6fa.

После того как текущий поток получит состояние синхронизации и выполнит соответствующую логику, ему необходимо освободить состояние синхронизации, чтобы последующие узлы могли продолжать получать состояние синхронизации. вызовом синхронизатораrelease(int arg)метод может освободить состояние синхронизации, метод вПосле сброса состояния синхронизации он разбудит свой узел-преемник.(что, в свою очередь, заставляет узел-преемник повторить попытку получить состояние синхронизации). Синхронизаторы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;
}

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    // 如果当前节点的状态小于0，那么用CAS设置成0
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    
    // 获取当前节点的后继节点
    Node s = node.next;
    // 如果后继节点为空 或者 后继节点的状态 > 0 (为取消状态)
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 从尾结点向前查找状态不为取消的可用节点
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        // 唤醒后继节点
        LockSupport.unpark(s.thread);
}

Когда этот метод выполняется, он разбудит поток узла-преемника головного узла,unparkSuccessor(Node node)использование методаLockSupportчтобы разбудить поток, который находится в состоянии ожидания.

Суммировать

При получении синхронного состояния синхронизатор поддерживает синхронный очередь, получая резьбу, что состояние не удалось добавить в очередь и вращаться в очереди; условие для удаления очереди (или STOPS SPIN) - это узел переднего привода Узел головы и успешное Получить синхронное состояние. Синхронизаторы вызовы, когда выделяются в состоянии синхронизацииtryRelease(int arg)Метод освобождает состояние синхронизации, а затем пробуждает узлы-преемники головного узла.

Получение и освобождение общего состояния синхронизации

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

Snipaste_2019-07-06_14-23-18.png

На приведенном выше рисунке в левой половине при совместном доступе к ресурсам разрешены другие общие доступы, а монопольный доступ заблокирован. Когда правая половина имеет монопольный доступ к ресурсу, другие доступы одновременно блокируются.

вызовом синхронизатораacquireShared(int arg)Метод может совместно использовать состояние синхронизации, и код метода показан ниже.

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

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);

    // propagate如果>0，说明我这次获取共享锁成功后，还有剩余共享锁可以获取
    // 如果=0，说明我这次获取共享锁成功后，没有共享锁可以获取

    /*
         如果propagate > 0，说明还有剩余共享锁可以获取，那么说明后继节点需要被唤醒。
         如果propagate = 0，说明没有剩余共享锁可以获取了，按理说不需要唤醒后继的。
         但是如果h.waitStatus < 0，这说明之前head的waitStatus < 0，不过在这之前waitStatus都被置为了0，
         只有一种可能会让waitStatus < 0
             由于doReleaseShared里的compareAndSetWaitStatus(h, 0, Node.PROPAGATE)的操作，
             有另一个线程在调用doReleaseShared才能造成，而这很可能是因为在中间状态时，又有人释放了共享锁
     */
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        // 如果当前节点的后继节点是共享类型或者当前节点没有后继节点，则进行唤醒
        // 这里可以理解为除非明确指明不需要唤醒（后继等待节点是独占类型），否则都要唤醒
        if (s == null || s.isShared())
            // 唤醒获取到共享锁节点的后继节点
            doReleaseShared();
    }
}

существуетacquireShared(int arg)метод, синхронизатор вызываетtryAcquireShared(int arg)метод пытается получить статус синхронизации,tryAcquireShared(int arg)Возвращаемое значение методаintТип, когда возвращаемое значение больше или равно 0, это означает, что состояние синхронизации может быть получено. Следовательно, в процессе спина совместного получения условием успешного достижения состояния синхронизации и выхода из спина являетсяtryAcquireShared(int arg)Метод Возвращаемое значение больше или равно 0.

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

Как и при монопольном, совместное получение также требует освобождения состояния синхронизации, вызываяreleaseShared(int arg)Метод может снять состояние синхронизации, код метода показан ниже.

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

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        // 如果头结点不为空 && 头结点不等于尾结点，说明存在有效的node节点
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // 如果头结点的状态为SIGNAL(-1)，说明存在需要唤醒的后继节点
            if (ws == Node.SIGNAL) {
                // 将头结点状态更新为0（初始值状态），此时头结点已经没用了
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // 唤醒后继节点
                unparkSuccessor(h);
            }
            // 如果头结点为初始值状态0，则设置为PROPAGATE(-3)，确保在释放同步状态时能通知后继节点
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        
        //如果头结点没有发生变化，表示设置完成，退出循环
        //如果头结点发生变化，比如说其他线程获取到了锁，为了使自己的唤醒动作可以传递，必须进行重试
        if (h == head)                   // loop if head changed
            break;
    }
}

После того как этот метод отключит состояние синхронизации, он разбудит последующие узлы в состоянии ожидания. Для параллельных компонентов, которые могут поддерживать одновременный доступ нескольких потоков (например,Semaphore), основное отличие между ним и монопольным типом заключается в том, чтоtryReleaseShared(int arg)Метод должен обеспечить, чтобы состояние синхронизации (или количество ресурсов) - это безопасно, обычно, обычно через петли и CAS, потому что операция выпуска состояния синхронизации будет приходить из нескольких нитей одновременно.

Уведомление:setHeadAndPropagateМетод указывает, что поток в очереди ожидания успешно получает разделяемую блокировку, в это время ему нужно разбудить разделяемый узел позади него, но когда он проходитreleaseSharedспособ снятия общей блокировки, затем подождитеПотоки с эксклюзивными блокировками и общими блокировками могут быть разбуженыПопробуй получить.

Суммировать

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

Суммировать

Разница между эксклюзивными блокировками и общими блокировками на исходном уровне заключается в следующем:

Эксклюзивные блокировки вызываются только текущей блокировкойreleaseПосле того как метод снимает блокировку, он может разбудить узел-преемник очереди ожидания CLH.

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

Категории