Иллюстрация общей блокировки AQS для семафора Semaphore — нечестный режим

Java

вводить

Ранее мы объясняли монопольную блокировку AQS, а в этой главе в основном объясняется общая блокировка AQS.SemaphoreСемафор для объяснения, я полагаю, что после прочтения содержания этой главы учащиеся смогут понять режим совместного использования AQS,SemaphoreСемафор обеспечивает количество одновременных доступов, используемых для управления ресурсом, то есть он будет поддерживать лицензию.Перед доступом к ресурсу необходимо подать заявку на получение лицензии.После успешной подачи заявки на лицензию вы можете получить к нему доступ. При подаче заявки на доступ к ресурсу лицензия получена.Если лицензия выдана, доступ к ресурсу может быть выполнен.При этом лицензия, выданная лицензионным центром, будет увеличиваться.После того, как поток, обращающийся к ресурсу, освобождает ресурс , использование лицензии уменьшится.

пример

public class SemaphoreDemo {
    private static final Semaphore semaphore = new Semaphore(3);
    private static final AtomicInteger atomicInteger = new AtomicInteger();

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        for (int i = 0; i < 10; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + "开始执行");
                    Thread.sleep(100);
                    semaphore.release();
                    System.out.println(Thread.currentThread().getName() + "执行完毕");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            });
        }
    }
}

Результаты приведены ниже:

pool-1-thread-1开始执行
pool-1-thread-2开始执行
pool-1-thread-3开始执行
pool-1-thread-1执行完毕
pool-1-thread-4开始执行
pool-1-thread-3执行完毕
pool-1-thread-2执行完毕
pool-1-thread-5开始执行
pool-1-thread-6开始执行
pool-1-thread-4执行完毕
pool-1-thread-7开始执行
pool-1-thread-6执行完毕
pool-1-thread-9开始执行
pool-1-thread-8开始执行
pool-1-thread-5执行完毕
pool-1-thread-7执行完毕
pool-1-thread-10开始执行
pool-1-thread-8执行完毕
pool-1-thread-9执行完毕
pool-1-thread-10执行完毕

Набрать примерSemaphoreЗавесаsemaphore.acquire();Подайте заявку на семафор, если приложение успешно, выполните следующую логику и передайте ее позже semaphore.release();Освободить занятость семафор, то естьsemaphore.acquiceПолучите разрешение от семафора, выполните следующую инструкцию, если разрешение передано, подождите, если разрешения нет, затем передайтеsemaphore.release();Возвращая разрешение, выходные данные в приведенном выше примере также ясно показывают, что только три потока могут получить доступ к ресурсу одновременно.

Объясните с помощью рисунков:

При первой инициализации на следующем рисунке показано, что размещаются 3 лицензии, как показано на следующем рисунке:Левая сторона — это поток, средняя — файл ресурсов, к которому нужно получить доступ, а правая сторона — пул лицензий. можно получить с семафора, то Вы можете получить доступ к ресурсу.Если лицензия не может быть выдана, значит, нужно подождать.Разберем операцию:Первый нить 1 Для доступа к ресурсам при доступе к доступу без разрешения не выполняется, требуется до приложения лицензии, после успешного приложения может получить доступ к ресурсам, то разрешение было бы сократить объем сигнала, посмотрите на изображение выше только двух лицензий будет иметь доступ к четыре потока, если нить находится в состоянии ожидания, как показано ниже:Как показано на рисунке выше, два средних потока упрощают процесс приложения и непосредственно имеют права доступа к разрешениям.Если приложение для получения разрешения не удается, оно будет ждать, пока другие потоки вернут разрешение, и обнаружит, что поток 4 ожидает разрешения. Когда другие потоки вызываютreleaseПосле метода лицензия будет возвращена семафору, как показано на следующем рисунке:Это когда поток 4 обнаружит, что уже есть доступные лицензии, он не будет ждать лицензии для быстрого доступа к ресурсам, и я не буду больше рисовать здесь картинки.Как и выше, примерно мы уже узнали о содержимом семафора. Далее мы проанализируем исходный код.

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

Введение в метод семафора

Основной метод семафора имеет следующее:

имя метода описывать
acquire() Попытки получить разрешение на доступ, если нет, поток ожидает, пока поток не освободит разрешение или когда поток будет прерван.
acquire(int permits) Попытки получить разрешения, если нет, поток ожидает, пока поток не освободит разрешения или когда поток будет прерван.
acquireUninterruptibly() Попытка получить разрешение на доступ, если его невозможно получить, поток ждет, пока поток выпустит разрешение, но не отвечает на запрос прерывания.
acquireUninterruptibly(int permits) Попытка получить разрешения, если это невозможно, поток ждет, пока поток освободит разрешения, но не отвечает на запрос прерывания.
release() Он используется для освобождения разрешения после окончания доступа к ресурсу потока, чтобы другие потоки, ожидающие разрешения, могли получить доступ к ресурсу.
release(int permits) Он используется для освобождения разрешения на разрешение после того, как поток обращается к ресурсу, чтобы другие потоки, ожидающие разрешения, могли получить доступ к ресурсу.
tryAcquire() Попытка получить разрешение, вернуть true, если разрешение успешно, вернуть fasle, если нет, не будет ждать, вернуться немедленно
tryAcquire(int permits) Попытка получить разрешения, вернуть True, если разрешение успешно, вернуть FASLE, если он не удается, он не будет ждать, вернуться немедленно
tryAcquire(int permits, long timeout, TimeUnit unit) Попытаться получить разрешения в течение указанного времени, вернуть false, если разрешение не получено в течение указанного времени, в противном случае вернуть true
tryAcquire(long timeout, TimeUnit unit) Попытаться получить лицензию в течение указанного времени, вернуть false, если разрешение не получено в течение указанного времени, в противном случае вернуть true
доступные разрешения(): Количество доступных на данный момент лицензий

получить семафор

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

public class SemaphoreDemo {
    private static final Semaphore semaphore = new Semaphore(1);
    private static final AtomicInteger atomicInteger = new AtomicInteger();

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        for (int i = 0; i < 2; i++) {
            Thread.sleep(100);
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + "开始执行");
                    Thread.sleep(10);
                    System.out.println(Thread.currentThread().getName() + "执行完毕");
//                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            });
        }
    }
}

Для приведенного выше примера мы сначала инициализировали 1 семафор, а пул потоков отправил 2 задачи.В это время все также думали о рабочем результате.Результат выполнения состоит в том, что только 1 поток может завершить выполнение, а остальные потоки нужно дождаться операции.Поскольку семафор потребляется, вот вывод:

pool-1-thread-1开始执行
pool-1-thread-1执行完毕

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

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

мы видим, что он вызываетsync.acquireSharedInterruptibly(1)метод, этоsnycФактическиSemaphoreвнутренний классSyncобъект экземпляра, то возникает проблема, этоsyncКогда инициализируется переменная? На самом деле, когда мы инициализируемSemaphore, ужеsyncПеременная инициализирована, давайте посмотримSemaphoreКонструктор:

// 非公平模式
public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}
// fair=true为公平模式，false=非公平模式
public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
  1. Конструктор: число разрешений инициализированного семафора и нечестный режим
  2. Конструктор 2: если указанное значение равно true, принимается честный режим, если указанное значение равно false, принимается нечестный режим, а число инициализированных семафоров равно допустимому.
/**
 * 非公平模式
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;

    NonfairSync(int permits) {
        super(permits);
    }
		//实现AQS的tryAcquireShared方法，尝试获得锁。
    protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);
    }
}

/**
 * 公平模式
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = 2014338818796000944L;

    FairSync(int permits) {
        super(permits);
    }
		//实现AQS的tryAcquireShared方法，尝试获得锁。
    protected int tryAcquireShared(int acquires) {
        for (;;) {
            if (hasQueuedPredecessors())
                return -1;
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
}

мы можем узнатьSemaphoreПредусмотрены два режима механизмов блокировки, один честный режим, а другой недобросовестный режим На самом деле честный режим заключается в том, что если поток находится в ожидании в очереди, он сознательно встанет в очередь позже, в то время как недобросовестный режим отличается , Независимо от того, есть ли у вас поток в очереди, кто бы ни взял его первым, здесь мы обнаруживаем, что в приведенном выше примере операторSemaphoreНа самом деле нечестный режим используется по умолчанию.NonfairSync, указывает, что количество семафоров равно 1, на самом деле он внутреннеSyncпозвонил вAQSизsetStateМетод, устанавливающий состояние синхронизацииstateравно 1, как показано на следующем рисунке:

Далее мы возвращаемся кacquireметод, он вызывает sync.acquireSharedInterruptibly(1);, осторожные друзья найдутNonfairSyncи родительский классSyncтакого метода нет вAQSМетоды предоставлены, то мы смотрим на этот метод в конце концов, чтобы делать то, что? Содержание источника следующим образом:

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())								//如果线程被中断则抛出异常。
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)					//尝试获取信号量，如果获得信号量小于0，则代表获取失败则运行下面的语句。
        doAcquireSharedInterruptibly(arg);	//将当前线程节点加入到队列中，等到其他线程释放信号量。
}
  1. Во-первых, оцените, прерван ли поток.Если он прерван, он будет прерван соответственно, и будет сгенерировано InterruptedException.
  2. Попробуйте получить семафор. Если значение семафора уже меньше 0, это означает, что семафора для получения нет. Если получение не удалось, он будет запущенdoAcquireSharedInterruptiblyметод для приостановки текущего потока в ожидании освобождения семафора другими потоками.

Далее мы смотрим наtryAcquireSharedреализация метода,tryAcquireSharedЭтот методAQSПредоставляет способ достижения подкласса, он не реализует себя, но генерируется исключение, классы, которые его реализуют, должны бытьSemaphoreизSyncclass, мы обнаружили, что существует две реализации этого метода, в том числе нечестный режимNonfairSync, другой в честном режимеFairSync, поскольку в приведенном выше примере мы использовали нечестный режим, давайте посмотрим на нечестный режимtryAcquireSharedЛогика реализации:

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

Это(NonfairSyncВнутренний вызов родительского классаSyncизnonfairTryAcquireSharedметод, продолжайте копать глубже и посмотрите на этот метод:

final int nonfairTryAcquireShared(int acquires) {
  	//这里上来就是个死循环
    for (;;) {
      	//获取可用的信号量数量。
        int available = getState();
      	//剩余线程数，其实就是当前可用信号量数量-申请的信号量数量
        int remaining = available - acquires;
      	//1. 如果剩余信号量数量小于0，代表没有信号量可用
      	//2. 修改state成功，则代表申请信号量成功。
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

Если вы читали предыдущуюReentrantLockДрузья исходного кода найдут, что это правильноstateувеличивается, т. е. еслиstateЕсли для значения установлено значение, это означает, что поток уже получил блокировку, и другим потокам не разрешено получать текущую блокировку.Если текущий поток повторно получает блокировку, тоstateЗначение увеличится, что также является ключевой точкой блокировков ReentRant, а такжеSemaphoreРазница в том, что этоstateСостояние синхронизатора уменьшается, другими словами, сначала инициализируются несколько семафоров.Если оставшийся семафор меньше 0 при получении семафора, это означает, что семафор недоступен, и он вернется напрямую.Если семафор успешно получено,stateИзмените значение, вернитесь к приведенному выше примеру, мы только что проанализировали первый поток, первый поток получил семафор, в это время оставшийся семафор равен 0, он будетstateЗначение установлено на 0, после установки возвращается кacquireSharedInterruptiblyизif (tryAcquireShared(arg) < 0)В инструкции оператор if истинен и не входит внутрь оператора if.На данный момент ситуация с AQS показана на следующем рисунке:Первый поток работает, но он не выпускает количество, а затем второй поток будет повторяться, когда второй поток начнет работать.Сначала сначала сначала перейти к сигналу приложения сначала.Количество, конечно, когда он применяется для a семафор, открытиеstateУже стал 0, когда он выполненtryAcquireSharedПри получении семафора доступный семафор равен 0. Когда доступный семафор минус количество запрашиваемых семафоров на 1, оставшийся семафор становится -1, поэтому условие оператора if в это времяremaining < 0Если он удовлетворен, введите оператор if и верните оставшийся семафор - 1. В это время он перейдет к месту вызова, то есть к AQS.acquireSharedInterruptiblyметод, на этот раз найденный в операторе if (tryAcquireShared(arg) < 0) Возвращаемый результат равен -1, и он введет оператор if для выполненияdoAcquireSharedInterruptiblyметод, основная операция этого метода заключается в том, чтобы поместить текущий поток в очередь ожидания, дождаться, пока другие потоки освободит семафор, а затем медленно проанализировать внутренний исходный код.AQS->doAcquireSharedInterruptibly:

/**
 * Acquires in shared interruptible mode.
 * @param arg the acquire argument
 */
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);				//将节点添加到队列中
    boolean failed = true;													//失败标志位，如果出现中断，或者异常抛出时会进行取消操作
    try {
        for (;;) {
            final Node p = node.predecessor();		  //获取当前节点的前节点
            if (p == head) {												//如果当前节点的前节点为头节点
                int r = tryAcquireShared(arg);			//尝试获得锁
                if (r >= 0) {												//如果可以获得信号量
                    setHeadAndPropagate(node, r);		//设置当前节点为头节点
                    p.next = null; // help GC				//帮助GC回收
                    failed = false;									//设置失败为false，也就是正常获取
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

В это время, когда наша программа запускается наaddWaiter, давайте посмотрим на ситуацию с очередью в это время:Содержание приведенного выше рисунка состоит в том, что когда второй поток получает семафор, получение завершается ошибкой, и он присоединяется к очереди.Если очередь пуста, когда он впервые присоединяется к очереди, он сначала создаст новый узел Node.Ref-405является головным узлом, а затем указывает узел текущего потока на головной узел, который является содержимым, показанным на рисунке выше, потому что здесьaddWaiterЭксклюзивные блокировки для внутреннего кода и ранее проанализированного AQS (т.ReentrantLockИсходный код) проанализирован, поэтому повторяться здесь не буду.После добавления потока в очередь ожидания он входит в бесконечный цикл for.Сначала доходит до получения головного узла текущего узла, т.е. картинка выше.Ref-405, а затем судите, будь то головной узел. Контент здесь на самом деле, чтобы попытаться выпустить семафор, чтобы увидеть, выделяется ли семафор. Если остальное количество возвращенных семафоров больше или равно 0, это означает, что SEMAPHORE SCRAMBLE успешно. Узел должен быть обработан, но в нашем примере, когдаtryAcquireSharedКогда возвращаемое значение равно -1, поэтому получаем количество семафоров, не будет вводиться содержимое, но мы сначала проведем анализ и анализ здесь.setHeadAndPropagate, заложив основу для будущего:

//从方法中也可以看出来大致意思是设置头节点，并且根据条件是否唤醒后继节点。
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // 记录一下原来的head头节点。
    setHead(node); // 设置新的节点设置为头节点。
    if (propagate > 0 || 			//如果有信号量
        	h == null || 				//头节点为空情况下
        	h.waitStatus < 0 || //如果原来的头结点的状态是负数，这里指的是SIGNAL和PROPAGATE两个状态
        	(h = head) == null || // 重新阅读头节点防止额外的竞争。
        	h.waitStatus < 0) {		//如果原来的头结点的状态是负数，这里指的是SIGNAL和PROPAGATE两个状态
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}
  1. Определить, есть ли еще семафор, и если да, то освободить поток, ожидающий в очереди.
  2. Если состояние исходного головного узла отрицательное, это относится к двум состояниям SIGNAL и PROPAGATE.
  3. Повторное чтение головного узла предотвращает дополнительную конкуренцию.

Когда он не войдет в метод setHeadAndPropagate, он перейдет к следующим шагам:

  if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();

shouldParkAfterFailedAcquireИзмените головной узел наSIGNAL,parkAndCheckInterruptЗаблокируйте поток, запустите здесь, поток приостановлен, ожидая пробуждения других потоков, статус очереди следующий:

семафор выпуска

когда мы звонимsemaphore.release()При освобождении семафора он фактически вызываетAbstractQueuedSynchronizerсерединаreleaseShared(int arg)метод, давайте взглянем на содержимое исходного кода:

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

Затем проанализируйте AQS вreleaseSharedметод:

public final boolean releaseShared(int arg) {
  	// 调用Semaphore实现的tryReleaseShared方法。
    if (tryReleaseShared(arg)) {
      	// 唤醒后记节点
        doReleaseShared();
        return true;
    }
    return false;
}

Сначала попытайтесь освободить семафор.Если семафор был освобожден успешно, вызовите doReleaseShared, чтобы разбудить ожидающий узел, и сообщите ожидающему узлу в очереди, что семафор готов к получению.

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
      	// 获取当前的state值
        int current = getState();
      	// 将当前的state值添加releases个信号量
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
      	// cas修改state值
        if (compareAndSetState(current, next))
            return true;
    }
}

На самом деле, наиболее важным методом являетсяdoReleaseSharedметод, давайте посмотрим на исходный код:

/**
 * 唤醒队列中的节点，以及修改头结点的waitStatus状态为PROPAGATE
 * 1. 如果头节点等待状态为SIGNAL，则将头节点状态设为0，并唤醒后继节点
 * 2. 如果头节点等待状态为0，则将头节点状态设为PROPAGATE，保证唤醒能够正常传播下去。
 */
private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
          	// 如果头结点的状态为SIGNAL则进行唤醒操作。
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
          	// 如果头节点状态为0，则将头节点状态修改为PROPAGATE，至于为什么会变成0，为什么要有PROPAGATE？，请看下文。
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

Значение статуса PROPAGATE

У вас может возникнуть вопрос, почему бы не распространить > 0 напрямую, а затем напрямую разбудить следующий узел? Здесь я процитирую ошибку в предыдущей версии, чтобы проиллюстрировать:BUG-6801020

Согласно описанию в ОШИБКЕ, затронутые номера версий — JDK 6u11 и 6u17.Код, упомянутый в ОШИБКЕ для воспроизведения ошибки, выглядит следующим образом:

import java.util.concurrent.Semaphore;

public class TestSemaphore {

   private static Semaphore sem = new Semaphore(0);

   private static class Thread1 extends Thread {
       @Override
       public void run() {
           sem.acquireUninterruptibly();
       }
   }

   private static class Thread2 extends Thread {
       @Override
       public void run() {
           sem.release();
       }
   }

   public static void main(String[] args) throws InterruptedException {
       for (int i = 0; i < 10000000; i++) {
           Thread t1 = new Thread1();
           Thread t2 = new Thread1();
           Thread t3 = new Thread2();
           Thread t4 = new Thread2();
           t1.start();
           t2.start();
           t3.start();
           t4.start();
           t1.join();
           t2.join();
           t3.join();
           t4.join();
           System.out.println(i);
       }
   }
}

Затем взгляните на затронутые номера версий.setHeadAndPropagateиreleaseSharedИсходный код двух методов выглядит следующим образом:

private void setHeadAndPropagate(Node node, int propagate) {
    setHead(node);
  	// 这里是区别点，他这里直接是比较的信号量如果存在，并且当前节点的等待状态不等于0，才会去唤醒下一个线程。
    if (propagate > 0 && node.waitStatus != 0) {
        /*
         * Don't bother fully figuring out successor.  If it
         * looks null, call unparkSuccessor anyway to be safe.
         */
        Node s = node.next;
        if (s == null || s.isShared())
            unparkSuccessor(node);
    }
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

Найдено в JDK версии JDK 6u11,6u17 нетPROPAGATEЭто состояние было введено в более поздних версиях для решения проблемы зависания потока, вызванной одновременным выпуском в общем режиме.После выполнения приведенного выше примера в течение определенного периода времени иногда происходит зависание потока. В приведенном выше примере инициализируются четыре потока, а семафор при инициализации равен 0. Потоки t1 и t2 должны получить семафор, а потоки t3 и t4 должны освободить семафор. Предполагая экстремальную ситуацию, t1 и t2 добавляются в очередь, как показано на следующем рисунке:

  1. В момент времени t1 поток t3 вызываетreleaseSharedметод, который вызоветunparkSuccessor, этот метод используется для уведомления ожидающего потока.В это время состояние ожидания в голове изменяется с -1 на 0, а затем пробуждается поток t1.В это время семафор равен 1.

  1. Поскольку поток t1 только что проснулся, его головной узел еще не переключился, потому что в приведенном выше примере мы можем ждать, пока поток ожидаетdoAcquireSharedInterruptiblyВ методе, когда поток просыпается, он также выполняется из этого метода.Когда поток t1 пытается получить семафор, он обнаруживает, что семафор можно получить.tryAcquireSharedВозвращаемое значение равно 0, так как семафор потребляется, и в это время текущий поток не продолжает работу, а выполняет переключение потоков.В это время состояние потока следующее:

  1. В этот момент переключитесь на поток t4, и вызовы t4releaseShared, в это время состояние ожидания = 0 головного узла напрямую возвращает false, и unparkSuccessor не вызывается, но в это время семафор становится равным 1.

  1. В это время пробуждается поток t1 и продолжает выполнять головной узел, указывающий на Ref-505, а семафор в это время только один.Он потребляет семафор сам.Хотя сейчас состояние=1, мы можем видеть семафор при переключении потока.состояние=0, поэтому после переключения потока обратно его распространение=0, вызовsetHeadAndPropagateПри вызове метода он не вошел внутрь оператора if, поэтому поток t2 не был разбужен, что привело к зависанию основного потока.

в jdk1.8setHeadAndPropagateне звонит напрямуюunparkSuccessorметод, но измените вызовdoReleaseSharedметод, давайте посмотрим, в чем разница между этим методом и описанной выше ошибкой:

/**
 * 1.唤醒队列中的节点
 * 2.如果队列头节点waitStatus=0，则将当前head头节点修改为PROPAGATE(-3)状态
 */
private void doReleaseShared() {
    /*
     * Ensure that a release propagates, even if there are other
     * in-progress acquires/releases.  This proceeds in the usual
     * way of trying to unparkSuccessor of head if it needs
     * signal. But if it does not, status is set to PROPAGATE to
     * ensure that upon release, propagation continues.
     * Additionally, we must loop in case a new node is added
     * while we are doing this. Also, unlike other uses of
     * unparkSuccessor, we need to know if CAS to reset status
     * fails, if so rechecking.
     */
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

Здесь мы возвращаемся к третьему шагу выше, переключаемся на поток t4 в это время, t4 вызываетreleaseShared, в это время состояние ожидания головного узла равно 0, и он возвращает false напрямую, и unparkSuccessor не вызывается, но семафор в это время становится равным 1, а состояние ожидания головного узла изменяется на -3.

13.png

Вернитесь к четвертому шагу выше: в это время пробуждается поток t1, и выполнение продолжает указывать головной узел наRef-505, а семафор на тот момент был только 1, он сам потреблял семафор, хотя сейчас состояние=1, мы можем видеть семафор при переключении потока.state=0, поэтому после того, как поток переключится обратно, егоpropagate=0,перечислитьsetHeadAndPropagateметод, состояние головного узла в это времяPROPAGATE(-3), введет оператор if для выполненияdoReleaseSharedметод, разбудите поток t2 в это время.

заключительные замечания

Так как статья писалась долго, а в середине был поиск работы, задержка была немного долгой, если есть какие-то ошибки, прошу меня поправить, а я со временем внесу исправления. Нелегко написать статью, и друзья, которые прочитали ее здесь, все хорошие.Поскольку статья немного длинная, в ней нет значка, обозначающего выпуск семафора, но процесс выпуска примерно объясняется позже, поэтому Я не буду здесь вдаваться в подробности. Благодарность

Друзья, которым это нравится, могут следить за моей публичной учетной записью WeChat и время от времени публиковать статьи.

123.png

Категории