Java concurrency — применение и принцип блокировок, этой статьи достаточно

задняя часть

Анализируя принцип блокировки, сосредоточьтесь на исходном коде ReentrantLock и ReentrantReadWriteLock и получите более глубокое понимание AQS за счет реализации блокировок. Эта статья очень длинная, не рассчитывайте понять принцип блокировки за 10 минут прочтения. Чтобы полностью понять эти знания, необходимо объединить исходный код с повторным опытом. Синхронизировано с моим личным блогом:Temple Network.GitHub.IO/2019/12/12/…

Задайте вопрос

В начале этой статьи мы сначала задаем вопросы, все наши исследования направлены на решение этих вопросов.

  • В чем разница между честным и нечестным режимом ReentrantLock? Как они?
  • Что такое реентерабельность? Как ReentrantLock обеспечивает повторный вход?
  • Чем ReentrantLock отличается от Synronized?
  • Какова роль состояния? Каков принцип ее реализации? В чем сходство и различие между Condition и Object.wait, Object.notify?
  • Как AQS достигает вращения? Потоки в AQS постоянно вращаются?
  • В чем разница между ReentrantReadWriteLock и ReentrantLock?
  • Как ReentrantReadWriteLock обеспечивает неблокирующее чтение?
  • Как AQS реализует монопольные и общие блокировки?

Основы блокировки

Прежде чем анализировать исходный код, сначала объясните основы, связанные с блокировкой.

Спинлок и мьютекс

Обычно существует два способа решения проблемы логической непротиворечивости многопоточных общих ресурсов:

  • Блокировка мьютекса: когда он обнаруживает, что ресурс занят, он блокирует себя до тех пор, пока ресурс не будет освобожден, а затем снова пытается его получить. Блокировки мьютексов влекут за собой накладные расходы, такие как переключение контекста потока и отправка сигналов, и теоретически производительность ниже, чем у спин-блокировок.
  • Спин-блокировка: когда ресурс оказывается занятым, он продолжает пытаться получить блокировку, поэтому он называется «спин». Спин-блокировка обнаружит флаг блокировки в бесконечном цикле, который занимает ЦП в течение периода, и когда конкуренция будет жесткой, флаг будет часто меняться, что вызовет высокочастотную синхронизацию кэша. Поэтому он подходит для ситуаций, когда время удержания блокировки относительно короткое.

Между этими двумя методами нет никакой разницы, только подходят ли они для текущей сцены.

Но если конкуренция очень высока, использование спин-блокировок может создать дополнительные проблемы:

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

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

замок CLH

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

Следовательно, должен быть способ заставить поток выполнения больше не вращаться вокруг одной и той же общей переменной и избегать чрезмерно частых операций синхронизации кэша. тогдаMCSиCLHЗамок появился на свет.Поскольку MCS и CLH очень похожи, идея CLH в основном используется в Java, поэтому CMS здесь изучаться не будет.

Названия замков CLH произошли от инициалов имени изобретателя:

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

Справедливый и несправедливый режим

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

Определение **FairSync** заключается в том, что все потоки, конкурирующие за ресурсы, следуют принципу очередности и получают блокировки один за другим. Несправедливые блокировки (NonfairSync) допускают «прыжки из очереди».

В реальной жизни все насмехаются над поведением по сокращению очередей, но все полностью следуют принципу организации очередей, а эффективность (в компьютерном мире эффективность заключается в максимальном использовании вычислительной мощности ЦП с наименьшими ресурсами) действительно высока? Чтобы привести простой пример, есть 10 потоков, ожидающих ресурсов в справедливой очереди, но поскольку поток A имеет большое количество операций ввода-вывода, он занимает блокировку и не освобождает ее, а ЦП свободен. В этот момент ресурсы процессора не могут быть использованы в полной мере.

Затем, если прибывает поток B с более быстрым выполнением, а ресурс оказывается в неконкурентном состоянии, то он может напрямую «вскочить», чтобы прийти и выполниться первым, но поскольку поток B выполняется очень быстро, поток A не много страдать.Следовательно, недобросовестные блокировки могут максимально использовать вычислительные ресурсы системы, избавляя от необходимости пробуждения потока, переключения из пользовательского режима в режим ядра и других ресурсоемких операций, а также повышая эффективность системы в целом. точка зрения..

Однако на все вещи нужно смотреть диалектически. Например, если поток A является очень важным потоком, но из-за жесткой конкуренции за блокировки поток A никогда не сможет получить блокировку, что повлияет на некоторые ключевые функции, т.е.блокировка голодания.Следовательно, когда конкуренция за замки жесткая, недобросовестные замки будут голодать..

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

возвращающийся

Давайте сначала поговорим о сценариях, чтобы понять повторный вход. Следующий код в конечном итоге напечатает 100. При вызове test основной поток получает блокировку, но когда основной поток не снял блокировку, он снова получает блокировку из-за рекурсивного вызова.Поведение, при котором поток может получить одну и ту же блокировку несколько раз, называется повторным входом в блокировку.. Если он не реентерабельный, следующий код вызоветтупик.

И ReentrantLock, и synchronized являются реентерабельными блокировками..

public class TestClass {
    public static void main(String[] args) {
        AtomicInteger i = new AtomicInteger(0);
        test(i);
        System.out.println(i);
    }

    public static synchronized void test(AtomicInteger i) {
        if (i.get() < 100) {
            i.incrementAndGet();
            test(i);
        }
    }
}

##结果 100

оптимистическая блокировка и пессимистическая блокировка

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

Synchronized и ReentrantLock в Java реализованы по идее пессимистических блокировок.Блокировки таблицы и блокировки строк в базе данных также являются реализациями пессимистических блокировок.

Яваjava.util.concurrent.atomicРазличные атомарные классы в пакете представляют собой реализации CAS, использующие оптимистическую блокировку.

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

Реализация оптимистической блокировки обычно осуществляется черезМеханизм номера версииилиCAS-алгоритмвыполнить.

Механизм номера версии

Как правило, в данные добавляютversionПоле, которое показывает, сколько раз поле было изменено.Классические реализации включают в себя оптимистичный дизайн блокировки базы данных, механизм _version ElasticSearch и т. д. В виде следующего псевдокода объясните идею номера версии.

var user = select user.age,user.id,user.version from t_user where id = 1001;
user.age++;
int oldVersion = user.version;
user.version++;
int result = update t_user set user.age=#{user.age},version=#{user.version} where id = 1001 and version=#{oldVersion};
if(result == 1) {
  return 200;
} else {
  while(true){
	  //在执行一次如上逻辑    
    if(result == 1) {
      break;
    }
  }
}

CAS-механизм

По конкретным вопросам CAS см.эта статья, автор написал более подробно, и пример ABA тоже взят из его поста в блоге.

КАСсравнить и поменять местами, является хорошо известным алгоритмом без блокировки. Алгоритм CAS включает три операнда

  • Значение памяти V, которое необходимо прочитать и записать
  • Значение A для сравнения
  • Новое значение B, которое нужно записать

CAS атомарно обновляет значение V новым значением B тогда и только тогда, когда значение V равно A, в противном случае он ничего не делает (Сравнение и замена являются атомарными операциями), постоянно повторяя попытку через вращение.

Но механизм CAS будет существоватьАВА-проблема, приведите пример вывода денег из жизни.

В многопоточном сценарииCASПоявитсяABAВопрос, вот простой научно-популярный вопрос по проблеме ABA.Например, есть два потока, которые выполняют CAS-операции над одним и тем же значением (начальное значение A) одновременно. Три потока выглядят следующим образом

  1. Поток 1, ожидаемое значение — A, а обновляемое значение — B.
  2. Поток 2, ожидаемое значение — A, а обновляемое значение — B.

нить1Упреждающий доступ к срезам процессорного времени, в то время как потоки2Тема заблокирована по другим причинам.1Сравните значение с ожидаемым значением A, найдите его равным, затем обновите значение до B, а затем в это времяпоток появился3, ожидаемое значение — B, а обновляемое значение — A., значение потока 3 сравнивается с ожидаемым значением B, и если оно оказывается равным, значение обновляется до A. В это время поток2Восстановление после блокировки и получение кванта процессорного времени, поток2Значение сравнивается с ожидаемым значением A, и если оно оказывается равным, значение обновляется до B, хотя поток2также завершает операцию, но поток2Не зная, что значение прошлоA->B->Aпроцесс изменения.

ABAВред, причиняемый проблемой: Сяо Мин снял в банкомате 50 юаней.Из-за проблемы с банкоматом было два потока, и баланс изменился со 100 на 50 одновременно. Поток 1 (банкомат): получить текущее значение 100, ожидать обновления до 50, Поток 2 (ATM): получить текущее значение 100, ожидать обновления до 50, Поток 1 успешно выполнен, а поток 2 по какой-то причине заблокирован.В это время кто-то переводит 50 Сяо Мину поток 3 (по умолчанию): получить текущее значение 50, ожидать обновления до 100, В это время поток 3 успешно выполняется, и баланс становится равным 100. Поток 2 восстанавливается из блока и получает 100. После сравнения он продолжает обновлять баланс до 50! ! ! На этом этапе вы можете видеть, что фактический баланс должен быть 100 (100-50+50), но на самом деле он становится 50 (100-50+50-50), что является успешной отправкой, вызванной проблемой ABA.

Решение: добавьте номер версии перед переменной, и переменная будет обновляться каждый раз при обновлении переменной.номер версии+1,СейчасA->B->Aэто становится1A->2B->3A.

synchronized

Synchronized — это встроенная блокировка в Java, а конкретная реализация — в JVM. Synchronized может изменять блоки кода, изменять методы и т. д. Поскольку его использование слишком простое, оно не будет повторяться в этой статье.

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

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

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

На вопрос: Чем ReentrantLock отличается от Synronized? Ответ:

До версии JDK1.6, поскольку смещенные блокировки и облегченные блокировки не были введены, реализация синхронизированной виртуальной машины JVM была относительно тяжелой, что приводило к определенным проблемам в ее производительности.Производительность ReentrantLock лучше, чем синхронизированная. Однако с непрерывной оптимизацией синхронизированного с помощью JVM после jdk1.6 разрыв между ними становится все меньше и меньше, и базовая реализация синхронизированного в основном полагается наLock-FreeОчередь, основная идеяблокировка после отжима,Продолжать борьбу за блокировки после переключения конкуренции,Слегка жертвует справедливостью, но получает высокую пропускную способность, разница в производительности незначительна. Так что в целом производительность не самый большой разрыв между ними. Настоящая разница заключается в использовании двух: синхронизация очень жесткая и может обеспечить реализацию недобросовестных монопольных блокировок только для одного блока кода или одного метода. По сравнению с синхронизированным, ReentrantLock очень гибок, он может блокировать как честные, так и нечестные блокировки.newCondition(), чтобы реализовать блокирующую очередь блокировок, но гибкость обеспечивает более высокий порог для использования, и легко вызвать взаимоблокировки, если использование не стандартизировано! Не стоит недооценивать тупиковую ситуацию, выход из тупиковой ситуации заключается в перезапуске процесса! Но при использовании синхронизированного нет необходимости рассматривать проблему взаимоблокировки.

ReentrantLock

Общий метод

  • void lock()получить замок

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

  • void lockInterruptibly()получить блокировку, если поток не прерван

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

  • boolean tryLock()Блокировка будет получена только в том случае, если ни один другой поток не удерживает блокировку в данный момент, немедленно вернитесь и вернитесьtrue, установите флаг на 1. Если текущий поток уже удерживает блокировку, немедленно вернуться, увеличить флаг и вернутьсяtrue. Если блокировка удерживается другим потоком, она немедленно возвращаетсяfalse.

  • boolean tryLock(long timeout, TimeUnit unit)Конкретная логика такая же, как и у tryLock, но есть дополнительное время ожидания: если поток прервется в течение периода ожидания, он также выброситInterruptedException

  • void unlock()Попытка снять блокировку, текущий поток владеет блокировкой, счетчик уменьшается на 1, и блокировка снимается до тех пор, пока счетчик не станет равным 0. Выдает, если текущий поток не удерживает блокировкуIllegalMonitorStateExceptionаномальный

  • Condition newCondition()Возвращает экземпляр Condition блокировки. Роль экземпляра Condition такая же, какObject.wait(),Object.notify(),Object.notifyAll()такой же. Если блокировка не удерживается ни одним потоком, вызовитеCondition.awart()илиCondition.signal, броситIllegalMonitorStateException. Если поток прерван, ожидание завершится броскомInterruptedException, бит флага прерывания потока будет очищен.

  • int getHoldCount(), количество очередей блокировок CLH

  • boolean isHeldByCurrentThread()Удерживается ли блокировка текущим потоком

  • boolean isLocked()Удерживается ли замок нитью

  • boolean isFair()Это честный замок

  • Thread getOwner()Возвращает поток, который в данный момент удерживает блокировку, или если блокировка не удерживается ни одним потокомnull

Принцип реализации справедливого и нечестного режима

Следующий код сохраняет только критический код.Syncпо наследствуAbstractQueuedSynchronizerполучить базовые возможности синхронного управления,NofairSyncиFairSyncУкажите квалификационный алгоритм для получения блокировок, реализуя метод шаблона AQS.

abstract static class Sync extends AbstractQueuedSynchronizer {
    
		...
    //如果线程可以获取锁，则返回true，否则立即返回false
    final boolean nonfairTryAcquire(int acquires) {
      final Thread current = Thread.currentThread();
      int c = getState();
      //如果状态为零，则设置ownerThread，返回true
      if (c == 0) {
        if (compareAndSetState(0, acquires)) {
          setExclusiveOwnerThread(current);
          return true;
        }
      }
      //如果状态不为零，判断当前线程是不是等于ownerThread,如果是则递增state，并立即返回true
      else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
          throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
      }
      //其他状态均返回false
      return false;
    }
	  ...
}

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

    final void lock() {
      //比较如果status=0则status=1，并设置ownerThread,通过这种方式使nonfairTryAcquire方法返回true
      if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
      else
        acquire(1);
    }

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

 static final class FairSync extends Sync {
     private static final long serialVersionUID = -3000897897090466540L;

     final void lock() {
       acquire(1);
     }

     protected final boolean tryAcquire(int acquires) {
       final Thread current = Thread.currentThread();
       int c = getState();
       if (c == 0) {
         //公平锁的实现中，比非公平锁多了一句hasQueuedPredecessors()，只有没有后继者并且status=0才可以有获取锁的资格，是非常之公平
         if (!hasQueuedPredecessors() &&
             compareAndSetState(0, acquires)) {
           setExclusiveOwnerThread(current);
           return true;
         }
       }
       else if (current == getExclusiveOwnerThread()) {
         int nextc = c + acquires;
         if (nextc < 0)
           throw new Error("Maximum lock count exceeded");
         setState(nextc);
         return true;
       }
       return false;
     }
}

Как добиться реентерабельности?

В предыдущей главе «справедливая блокировка» и «несправедливая блокировка» вы можете увидетьcurrent == getExclusiveOwnerThread(), status добавит полученное значение для увеличения и вернет true, поэтому после поступления того же потока он не будет добавлен в очередь синхронизации CLH.

Анализ принципа блокировки и разблокировки

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

  • tryAcquireПопытаться получить блокировку и вернуться напрямую, если блокировку можно получить.tryAcquire— это шаблонный метод, реализуемый классом, реализующим AQS. В ReentrantLock есть две реализации справедливой блокировки и нечестной блокировки, Конкретную реализацию см. в соответствующей главе.
  • Если обнаружено, что блокировка не может быть получена, вызовитеaddWaiterметод, положитьcurrentThreadупаковано вNodeи добавлен в очередь синхронизации. вот детальaddWaiter(Node.EXCLUSIVE), ReentrantLock — это идея эксклюзивной блокировки, поэтому нужно указать режим какэксклюзивный режим.
  • воплощать в жизньacquireQueuedметод, использующий метод вращения + блокировки для управления очередью синхронизации. Только в состоянии waitStatus=SIGNAL поток будет заблокирован, и каждый спин будет определять узел-предшественник текущего узла узла, независимо от того, является ли онheadУзел, если он является головным узлом и может получить блокировку, выскакивает из вращения.
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
    acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();
}

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
      node.prev = pred;
      if (compareAndSetTail(pred, node)) {
        pred.next = node;
        return node;
      }
    }
    enq(node);
    return node;
}

//独占模式的acquire方法
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
      boolean interrupted = false;
      //自旋+park
      for (;;) {
        final Node p = node.predecessor();
        //如果前置节点时head,注意此处还会再尝试一下能否获得到锁
        if (p == head && tryAcquire(arg)) {
          //这直接把node设置为head
          setHead(node);
          p.next = null; // help GC
          failed = false;
          return interrupted;
        }
        //如果前置节点不是head或者tryAcquuire返回false，则需要park节点
        if (shouldParkAfterFailedAcquire(p, node) &&
            parkAndCheckInterrupt())
          interrupted = true;
      }
    } finally {
      if (failed)
        cancelAcquire(node);
    }
}

//pred为前置节点，node为当前节点
//当获取锁失败时，检查和更新node.waitStatus，如果线程应该被阻塞，则返回true.
//这个方法是自旋锁中最重要的控制手段
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //获取到前置节点的waitStatus
    int ws = pred.waitStatus;
    //如果前置节点的状态为SIGNAL,则返回true，意味着当前节点node应该被park
    if (ws == Node.SIGNAL)
      return true;
    //如果前置节点被CANCEL的，则移除
    if (ws > 0) {
      do {
        node.prev = pred = pred.prev;
      } while (pred.waitStatus > 0);
      pred.next = node;
    } else {
      //原子的设置前置节点为SIGNAL
      compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

//阻塞线程并返回线程是否中断
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

protected final boolean tryRelease(int releases) {
    //重置标识位为0，这个很关键
    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;
}

//释放锁
public final boolean release(int arg) {
    //判定currentThread是否等于ownerThread,如果不相等则抛出IllegalMonitorStateException
    //然后，判断status属性是否等于0,如果等于0则返回true，同时清空ownerThread。代表资源已经被释放掉
    if (tryRelease(arg)) {
      Node h = head;
      if (h != null && h.waitStatus != 0)
        //unpark线程
        unparkSuccessor(h);
      return true;
    }
    return false;
}

Графический принцип конкуренции замков ReentrantLock

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

как показано на рисунке:

  • определениеReentrantLockПример, в данный момент head=null, tail=null, status=0
  • Поток А вызываетlock.lock(), так как поток A первым получает блокировку, статус = 1, в соответствии сacquireЛогика метода не устанавливает ни одного узла в очередь синхронизации, поэтому начало и конец по-прежнему равны нулю.
  • Поток B вызываетlock.lock(), поскольку на данный момент status=1, поэтомуtryAcquireВернуть ложь. запущенныйaddWariterЛогика, добавляются два узла Node и NodeB, а указатели начала и конца также указывают на соответствующий заголовок связанного списка и конец связанного списка.acquireQueuedЗапустите операцию отжима, первый отжим из-заNode.waitStatus != SIGNAL, поэтому после установки пре-узла waitStatus=SIGNAL продолжаем крутиться и снова определяемshouldParkAfterFailedAcquireметод, обнаружитьNode.waitStatus == SIGNAL, который заблокирует ThreadB.
  • Вызовы потока Clock.lock(), логика выполнения такая же, как и выше. Наконец, будет сформирована очередь синхронизации CLH, такая как NodeNodeBNodeC.
  • После завершения выполнения бизнес-логики потока A вызовитеlock.unlock(), который в конечном итоге вызоветreleaseМетод, установите статус = 0 и разбудите узел-преемник NodeB головы.После того, как NodeB просыпается, он продолжает вращаться, и, наконец, NodeB становится головным узлом, выходит из вращения и выполняет логику ThreadB .
  • Когда поток B завершит выполнение, вызовитеlock.lock(), в конечном итоге разбудит NodeC, и, наконец, все потоки будут выполнены.

AQS实现原理图解

Принципиальный анализ состояния

Создайте объект Condition для текущего объекта Lock со следующим кодом. Его эффект эквивалентенObject.wait()иObject.notify(), очередь условий представляет собой очередь FIFO, к которой можно получить доступsignalAllРазблокирует все потоки или черезsignalРазблокируйте темы по отдельности.

ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();

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

прежде всегоawaitЛогика:

  • Выбрасывает InterruptedException, если считается, что оно было прервано
  • Добавьте узел в очередь условий FIFO (очередь условий), состояние узла — СОСТОЯНИЕ, и очистите узел, состояние которого ОТМЕНЕНО, при добавлении узла
  • удалить узел из очереди синхронизации
  • Определить, находится ли узел в очереди на синхронизацию, заблокировать поток, только если его нет в очереди на синхронизацию
  • блочный узел
  • Когда узел получает сигнал, узел добавляется в очередь синхронизации, готовый получить блокировку.
  • Удалить все узлы, состояние которых ОТМЕНЕНО
  • Сообщить о прерывании. Если поток прервется до сигнала, будет выдано исключение прерывания. Если поток прервется после сигнала, он будет прерван снова, и вызывающий код сам обработает флаг прерывания.

ПослеsignalЛогика:

  • УдалитьfirstWaiter, пусто, если first.nextWaiter == nulllastWaiter
  • Преобразование узла из очереди условий -> очередь синхронизации
    • Состояние узла должно удовлетворять УСЛОВИЮ
    • разблокировать нить
//等待Conditon
public final void await() throws InterruptedException {
    //第一步，判断中断则抛出InterruptedException
    if (Thread.interrupted())
      throw new InterruptedException();
    //第二步，在FIFO条件队列（condition queue）增加节点，节点状态为CONDITION，在增加节点的同时清除掉状态为CANCELED的Node
    Node node = addConditionWaiter();
    //第三步,把该节点从同步队列(sync queue)中去除
    int savedState = fullyRelease(node);
		
    int interruptMode = 0;
    //第四步，判定节点是否在同步队列中，只有不在同步队列，才阻塞线程
    while (!isOnSyncQueue(node)) {
      //阻塞node
      LockSupport.park(this);
      if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
        break;
    }
    //第五步，当node被signal之后，把node加入到同步队列中，准备获取锁
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
      interruptMode = REINTERRUPT;
    //第六步，移除掉所有状态为CANCELED的节点
    if (node.nextWaiter != null) // clean up if cancelled
      unlinkCancelledWaiters();
  	//第七步，报告中断，如果在signal之前线程被中断则抛出中断异常，如果在signal之后线程被中断，则再次中断，有调用代码自行处理中断标识
    if (interruptMode != 0)
      reportInterruptAfterWait(interruptMode);
}
//新的waiter节点加入条件队列
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
      unlinkCancelledWaiters();
      t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
      firstWaiter = node;
    else
      t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

//轮询整个条件队列(condition queue)，去掉节点状态不是CONDITION的节点
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
      Node next = t.nextWaiter;
      //如果t的状态不是CONDITION,则把t节点从链表中摘除
      if (t.waitStatus != Node.CONDITION) {
        t.nextWaiter = null;
        if (trail == null)
          firstWaiter = next;
        else
          trail.nextWaiter = next;
        if (next == null)
          lastWaiter = trail;
      }
      else
        trail = t;
      t = next;
    }
}
//从同步队列中移除node
final int fullyRelease(Node node) {
    boolean failed = true;
    try {
      int savedState = getState();
      if (release(savedState)) {
        failed = false;
        return savedState;
      } else {
        throw new IllegalMonitorStateException();
      }
    } finally {
      if (failed)
        node.waitStatus = Node.CANCELLED;
    }
}
//判定node是否在同步队列中，条件队列的next和prev都一定是null,条件队列是nextWaiter的单向链表
final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
      return false;
    //如果node.next != null，说明存在后继节点，说明它一定在同步队列中
    if (node.next != null) // If has successor, it must be on queue
      return true;
    //从tail轮询同步队列，判定node是否在同步队列中
    return findNodeFromTail(node);
}

//检测中断，返回THROW_IE如果中断发生在signalled之前，返回REINTERRUPT如果中断发生在中断之后，返回0表示没有发生中断
private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
      (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
    0;
}

//wait结束后报告中断
private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
      throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
      selfInterrupt();
}

//释放Condition
public final void signal() {
    //模板方法，由实现AQS的实现类实现，在ReentrantLock中，判定current线程是否等于ownerThread
    if (!isHeldExclusively())
      throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
      doSignal(first);
}

private void doSignal(Node first) {
    do {
      if ( (firstWaiter = first.nextWaiter) == null)
        lastWaiter = null;
      first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}
//转换node从 条件队列-> 同步队列
final boolean transferForSignal(Node node) {  
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
      return false;
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
      //解除节点阻塞
      LockSupport.unpark(node.thread);
    return true;
}

Обработка прерывания по условию

При использовании Condition обязательно обратите внимание на обработку прерываний.Condition.await()иObject.wait()Напротив, последний поддерживает только прерываниеInterruptedExceptionисключение, в то время как первый бросаетInterruptedExceptionException, бит флага прерывания будет снова сброшен, а логика обработки прерывания по условию будет изменена.Если поток прервется до сигнала, будет сгенерировано исключение прерывания, если поток прервется после сигнала, он будет прерван еще раз, и вызывающий код сам обработает флаг прерывания.. Как показано в примере кода ниже.

определить три потокаt1,t2,t3,существуетt2воплощать в жизньsignal()До,t3поток выполненt1.interrupt(), исключение прерывания перехватывается в потоке t1. Если не реализованоt3нить и вt2воплощать в жизньsignal()тогда позвониt1.interrupt(), поток t1 не будет перехватывать исключение прерывания, ноThread.currentThread().isInterrupted() == true.

/** 在退出wait之后再次中断 */
private static final int REINTERRUPT =  1;
/** 在退出wait之后抛出InterruptedException异常 */
private static final int THROW_IE    = -1;

public static void main(String[] args) {
    ReentrantLock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    Thread t1 = new Thread(() -> {
      try {
        lock.lock();
        condition.await();
        System.out.println("aaa=" + Thread.currentThread().isInterrupted());
      } catch (InterruptedException e) {
        e.printStackTrace();
      } finally {
        lock.unlock();
      }

    });

    Thread t2 = new Thread(() -> {
      try {
        Thread.sleep(2000);
        lock.lock();
        condition.signal();
        //t1.interrupt();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (IllegalMonitorStateException e) {
        e.printStackTrace();
      } finally {
        lock.unlock();
      }

    });


    Thread t3 = new Thread(() -> {
      try {
        Thread.sleep(1000);
        t1.interrupt();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });

    t1.start();
    t2.start();
    t3.start();
}

Суммировать

Поскольку ReentrantLock — это самая простая реализация AQS, очень важно понимать ReentrantLock. Анализируя исходный код ReentrantLock, можно сделать следующие выводы:

  • ReentrantLock имеет два режима: честный и несправедливый.
  • ReentrantLock — это реентерабельная блокировка.
  • ReentrantLock – это эксклюзивная блокировка. Основная логика обеспечивается платформой AQS. Она просто резюмируется как очередь вращения + синхронизация для достижения эксклюзивной блокировки.
  • ReentrantLock включает ** очередь синхронизации (очередь синхронизации)иОчередь условий (очередь условий)** две очереди
  • Идентификационный статус блокировки AQS:statusПоле, узел очереди синхронизации является внутренним классомNode, состояние узла передается черезwaitStatusконтролировать, делитьсяSIGNAL,CANCELED,CONDITION,PROPAGATEчетыре состояния,nextWaiterпредставляет собой односвязный список условных очередей.
  • Условие реализовано путем блокировки потоков.Когда вся очередь условий будет выполнена, узел будет добавлен в очередь синхронизации.
  • AQS имеет ограниченное количество вращений, и когда waitStatus=-1 предыдущего узла заблокирует текущий узел. И каждый раз только следующий узел головного узла пробуждается для участия в блокировке соревнования, чтобы избежать слишком частого переключения контекста, вызванного пробуждением нескольких потоков одновременно.
  • ReentrantLock должен помнить о разблокировке, иначе возникнет взаимоблокировка.

ReentrantReadWriteLock

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

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

Обзор ReentrantReadWriteLock

Методы в ReadLock и WriteLock реализованы через класс Sync. Синхронизация является подклассом AQS, который, в свою очередь, выводит честные и несправедливые режимы.

Из метода Sync, который они вызывают, мы видим:ReadLock использует общий режим, WriteLock использует эксклюзивный режим..

Как один и тот же экземпляр AQS может одновременно использовать общий и монопольный режимы? ? ?

Суть AQS заключается во внутреннем состоянии атрибута: для монопольного режима обычно 0 означает, что блокировка может быть получена, 1 означает, что блокировка получена другими, а в реентерабельном исключении и совместно используемом режиме каждый поток может добавить или вычесть блокировку. То есть работа состояния в эксклюзивном режиме и в совместно используемом режиме совершенно различна Как состояние использует блокировку чтения-записи ReentrantReadWriteLock? Ответ состоит в том, чтобы разделить 32-битное целочисленное значение состояния на старшие 16 бит и младшие 16 бит, которые используются для разделяемого режима и эксклюзивного режима соответственно.

public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable {
    /** 内部类 读锁 */
    private final ReentrantReadWriteLock.ReadLock readerLock;
    /** 内部类 写锁 */
    private final ReentrantReadWriteLock.WriteLock writerLock;
    /** AQS 读写锁实现类 */
    final Sync sync;

    /**
     * 默认创建非公平锁
     */
    public ReentrantReadWriteLock() {
        this(false);
    }
		/**
		* 1、构建同步器：公平的&非公平的
		* 2、构建读锁和死锁实例
		*/
    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }
    
    public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
    public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }

    abstract static class Sync extends AbstractQueuedSynchronizer {}

    static final class NonfairSync extends Sync{}

    static final class FairSync extends Sync{}

    public static class ReadLock implements Lock, java.io.Serializable{}

		public static class WriteLock implements Lock, java.io.Serializable{}

}

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

получение блокировки чтения

В AQS, если возвращаемое значение метода tryAcquireShared(arg) меньше 0, это означает, что общая блокировка (блокировка чтения) не получена, а если больше 0, это означает, что она получена.

Общий режим AQS: метод tryAcquireShared используется не только в начале AcquireShared, это попытка, которая может дать сбой.Если это не удается, выполните следующий doAcquireShared, войдите в очередь блокировки, а затем дождитесь пробуждения узла-предшественника. После пробуждения tryAcquireShared по-прежнему будет вызываться для получения общей блокировки. Конечно, очень легко получить блокировку, повторив попытку после пробуждения, потому что другие узлы все еще находятся в состоянии блокировки, а в соревновании в этот момент участвует только пробужденный узел или другие новые потоки.

Итак, когда вы посмотрите на приведенный ниже код, представьте себе два сценария получения блокировок чтения: один для новичков и один для постановки в очередь.

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

//返回-1代表不能获取读锁
//返回1代表可以获取读锁
protected final int tryAcquireShared(int unused) {
    
    Thread current = Thread.currentThread();
    int c = getState();
    //如果独占锁的数量不等于零，说明有写锁，而且当前线程不等于ownerThread，说明不是重入线程，所以返回-1
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
      return -1;
    //获得共享锁数量
    int r = sharedCount(c);
    //如果读锁不应该阻塞且CAS更新成功
    //readerShouldBlock在公平模式和非公平模式有不同的实现
    //在公平模式，只要同步队列中有节点在排队，则新的节点就乖乖的去排队
    //在非公平模式，只有同步队列的head是写锁，才会去排队，否则不需要阻塞
    //这段代码中对firstReader、firstReaderHoldCount、readHolds等设定，都是出于性能考量，可以直接返回1
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
      //首次获取锁
      if (r == 0) {
        //出于性能考虑，可以
        firstReader = current;
        firstReaderHoldCount = 1;
      } else if (firstReader == current) {
        firstReaderHoldCount++;
      } else {
        //出于性能考虑
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != getThreadId(current))
          cachedHoldCounter = rh = readHolds.get();
        else if (rh.count == 0)
          readHolds.set(rh);
        rh.count++;
      }
      return 1;
    }
    return fullTryAcquireShared(current);
}

Оглядываясь назад на этот код, это условие суждения является ключевым моментом.if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARD_UNIT), есть три условия суждения:

  • readerShouldBlock()

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

      final boolean readerShouldBlock() {
  	return hasQueuedPredecessors();
}

public final boolean hasQueuedPredecessors() {
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    return h != t &&
      ((s = h.next) == null || s.thread != Thread.currentThread());
}

    • существуетNonFairSync, если узел-преемник головы является узлом потока блокировки записи, получение блокировки чтения блокируется, и сначала выполняется блокировка записи. Если head.next не является блокировкой записи, допускается конкуренция в нечестном режиме.

      final boolean readerShouldBlock() {
		return apparentlyFirstQueuedIsExclusive();
}

final boolean apparentlyFirstQueuedIsExclusive() {
  Node h, s;
  return (h = head) != null &&
    (s = h.next)  != null &&
    !s.isShared()         &&
    s.thread != null;
}

  • r < MAX_COUNT,static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;, MAX_COUNT обычно не достигается, а максимальное значение равно 65 535. Это условие можно игнорировать.

  • compareAndSetState(c, c + SHARD_UNIT), соревнование CAS терпит неудачу, это может быть другое соревнование блокировки чтения или может быть соревнование операции блокировки записи.

/**
* 这个方法是为了解决：
* 1、CAS竞争失败，因为经过readerShouldBlock方法的验证，线程不需要阻塞，如果仅仅是因为CAS竞争失败而导致进 * 入同步队列，就太可惜了。
* 2、处理重入
*
*/
final int fullTryAcquireShared(Thread current) {
    HoldCounter rh = null;
    //自旋
    for (;;) {
      //获取状态
      int c = getState();
      //获取写锁数量，如果此刻有线程获取到了写锁，则宣告读线程进入队列排队
      if (exclusiveCount(c) != 0) {
        if (getExclusiveOwnerThread() != current)
          return -1;
      //再次判定reader是否需要进入队列
      } else if (readerShouldBlock()) {
        if (firstReader == current) {
          // assert firstReaderHoldCount > 0;
        } else {
          //为了确保读锁重入操作能成功，而不是被塞到同步队列中等待
          if (rh == null) {
            rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current)) {
              rh = readHolds.get();
              if (rh.count == 0)
                readHolds.remove();
            }
          }
          if (rh.count == 0)
            return -1;
        }
      }
      //判定共享锁的数量，最大为65535
      if (sharedCount(c) == MAX_COUNT)
        throw new Error("Maximum lock count exceeded");
      //再次参加CAS竞争，如果还没有竞争到则继续自旋
      if (compareAndSetState(c, c + SHARED_UNIT)) {
        if (sharedCount(c) == 0) {
          firstReader = current;
          firstReaderHoldCount = 1;
        } else if (firstReader == current) {
          firstReaderHoldCount++;
        } else {
          if (rh == null)
            rh = cachedHoldCounter;
          if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
          else if (rh.count == 0)
            readHolds.set(rh);
          rh.count++;
          cachedHoldCounter = rh; // cache for release
        }
        return 1;
      }
    }
}

прочитать освобождение блокировки

Процесс снятия блокировки чтения относительно прост, главное уменьшить счетчик холдов на 1. Если он уменьшился до 0, удалить ThreadLocal.

Затем в цикле for уменьшите старшие 16 бит состояния на 1. Если обнаружено, что и блокировка чтения, и блокировка записи сняты, то разбудите следующий поток, который получает блокировку записи.

//共享模式release
public final boolean releaseShared(int arg) {
  if (tryReleaseShared(arg)) {
    doReleaseShared();
    return true;
  }
  return false;
}

protected final boolean tryReleaseShared(int unused) {
  Thread current = Thread.currentThread();
  if (firstReader == current) {
    // assert firstReaderHoldCount > 0;
    if (firstReaderHoldCount == 1)
      firstReader = null;
    else
      firstReaderHoldCount--;
  } else {
    HoldCounter rh = cachedHoldCounter;
    if (rh == null || rh.tid != getThreadId(current))
      rh = readHolds.get();
    int count = rh.count;
    if (count <= 1) {
      readHolds.remove();
      if (count <= 0)
        throw unmatchedUnlockException();
    }
    --rh.count;
  }
  for (;;) {
    int c = getState();
    int nextc = c - SHARED_UNIT;
    if (compareAndSetState(c, nextc))
      return nextc == 0;
  }
}

получение блокировки записи

Сначала поговорим о главном

1. Блокировка записи является эксклюзивной блокировкой.

2. Если блокировка чтения занята, при получении блокировки записи она войдет в очередь блокировки и будет ждать.

protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    //获取status值和写锁数量
    int c = getState();
    int w = exclusiveCount(c);
    if (c != 0) {
      //如果存在读锁，则写锁进入队列阻塞
      // (Note: if c != 0 and w == 0 then shared count != 0)
      if (w == 0 || current != getExclusiveOwnerThread())
        return false;
      if (w + exclusiveCount(acquires) > MAX_COUNT)
        throw new Error("Maximum lock count exceeded");
      // Reentrant acquire
      setState(c + acquires);
      return true;
    }
    //判定写锁是否应该阻塞，或者CAS静态失败，都会导致写锁阻塞
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
      return false;
    //设置独占锁ownerThread
    setExclusiveOwnerThread(current);
    return true;
}

static final class NonfairSync extends Sync {
    //非公平模式，多个写锁之间就靠CAS去抢锁了，抢不到进入队列中排队
    final boolean writerShouldBlock() {
      return false; 
    }
}

static final class FairSync extends Sync {
    //公平模式如果队列存在节点则进入队列排队
    final boolean writerShouldBlock() {
      return hasQueuedPredecessors();
    }
}

написать снятие блокировки

Освобождение блокировки записи очень просто.Освобождение монопольной блокировки является потокобезопасным, CAS не требуется, а состояние установлено на 1.

protected final boolean tryRelease(int releases) {
    if (!isHeldExclusively())
      throw new IllegalMonitorStateException();
    int nextc = getState() - releases;
    boolean free = exclusiveCount(nextc) == 0;
    if (free)
      setExclusiveOwnerThread(null);
    setState(nextc);
    return free;
}

заблокировать понижение версии

Дуг Ли не говорил, что блокировки записи — это нечто большее.передовой, если поток удерживает блокировку чтения, то получение блокировки записи также должно подождать.

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

Дуг Ли называет процесс получения блокировки чтения потоком, удерживающим блокировку записи, какЗаблокировать понижение версии. Таким образом, этот поток удерживает как блокировку записи, так и блокировку чтения.

но,блокировка обновленияэто невозможно. Если поток удерживает блокировку чтения, он не может получить блокировку записи, не сняв ее, потому что это произойдеттупик.

Примечание: Блокировки чтения-записи часто приводят к взаимоблокировкам в таких местах.Вы должны хорошо понимать эти идеи, прежде чем сможете легко их использовать.

Вернитесь назад и посмотрите на исходный код получения блокировки записи:

//锁不能升级的实现
protected final boolean tryAcquire(int acquires) {

    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c);
    if (c != 0) {
        // 看下这里返回 false 的情况：
        //   c != 0 && w == 0: 写锁可用，但是有线程持有读锁(也可能是自己持有)
        //   c != 0 && w !=0 && current != getExclusiveOwnerThread(): 其他线程持有写锁
        //   也就是说，只要有读锁或写锁被占用，这次就不能获取到写锁
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        ...
    }
    ...
}
//锁可以降级的实现
protected final int tryAcquireShared(int unused) {
  Thread current = Thread.currentThread();
  int c = getState();
  //重点！看这里，如果写锁数量不为0，代表有线程持有写锁。
  //但是！如果current == ownerThread ，则跳出这个判定，有机会获取读锁，这就是锁降级。
  //如果current != ownerThread，则读锁线程进入队列阻塞
  if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
    return -1;
  int r = sharedCount(c);
  if (!readerShouldBlock() &&
      r < MAX_COUNT &&
      compareAndSetState(c, c + SHARED_UNIT)) {
   ...
  }
  ...
}

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

Если поток а сначала получает блокировку записи, а затем блокировку чтения, то поток а не будет заблокирован и по-прежнему будет иметь возможность получить блокировку чтения.

код тупика

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

Ниже перечислены некоторые коды, которые могут вызывать взаимоблокировки.:

  • Случай 1: эскалация блокировки вызывает взаимоблокировку. После того, как поток получает блокировку чтения, он должен разблокировать блокировку чтения, прежде чем получить блокировку записи.

    public static void main(String[] args) {
    try {
      System.out.println("获取读锁");
      lock.readLock().lock();
      System.out.println("获取写锁");
      lock.writeLock().lock();
      lock.writeLock().unlock();
      System.out.println("释放写锁");
      lock.readLock().unlock();
    } catch (Exception e) {
      e.printStackTrace();
    } finally {

    }
}

  • Случай 2: дважды заблокировать, один раз разблокировать из-за невнимательности. Обязательно проверьте несколько раз, что блокировка и разблокировка сопряжены.

    public static void main(String[] args) {
    try {
        Random random = new Random();

        lock.writeLock().lock();
        lock.writeLock().lock();
        System.out.println("获取写锁");
        lock.readLock().lock();
        System.out.println("获取读锁");
        lock.readLock().unlock();
        System.out.println("释放读锁");

        lock.writeLock().unlock();

        new Thread(() -> {
          lock.readLock().lock();
          System.out.println("另外线程获取读锁");
          lock.readLock().unlock();
        }).start();
        System.out.println("释放写锁" + lock.writeLock().isHeldByCurrentThread());
    } catch (Exception e) {
      e.printStackTrace();
    } finally {

    }
}

Суммировать

1. Блокировки чтения-записи делятся наблокировка чтенияиблокировка записи, несколько потоков могут совместно использовать блокировки чтения, но только один поток может получить блокировки записи. После получения блокировки записи блокировка чтения ожидает снятия блокировки записи в очереди синхронизации. Это максимизирует производительность пропускной способности, обеспечивая согласованность данных.

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

3. Блокировка чтения реализована в общем режиме AQS, а блокировка записи реализована в монопольном режиме AQS. АКСstatus, делится на старшие 16 бит и младшие 16 бит, которые используются в совместно используемом режиме и эксклюзивном режиме соответственно.

4. Максимальное количество блокировок на чтение и запись — 65535. При превышении этого значения будет сброшеноError. Конечно, как правило, не превышается.

AbstractQueuedSynchronizer

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

Как самая основная структура в JUC, AQS может написать статью, просто рассказав о ней. В этой статье мы не собираемся подробно рассказывать об AQS, а лишь обобщаем и уточняем знания об AQS, описанные выше. Если вы можете понять ReentrantLock и ReentrantReadWriteLock, вы имеете определенное представление о сути AQS.

/**
*    * AQS是Node节点构成的同步队列
     *      +------+  prev +-----+       +-----+
     * head |      | <---- |     | <---- |     |  tail
     *      +------+       +-----+       +-----+
     * 
*/
public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
  /** 标识为共享模式 */
  static final Node SHARED = new Node();
  /** 表示为独占模式 */
  static final Node EXCLUSIVE = null;

  private transient volatile Node head;
  private transient volatile Node tail;

  /**
    * 同步位，可重入，state标识重入次数。为零时代表资源可以被获取
    * 读写锁时，把32位的int类型拆分出高16位和低16位，来区分读锁和写锁的状态
   */
  private volatile int state;
  
  static final class Node {
        /** 标识为共享模式，赋值给nextWaiter */
        static final Node SHARED = new Node();
        /** 标识为独占模式，赋值给nextWaiter */
        static final Node EXCLUSIVE = null;

        /** waitStatus状态>0是取消状态，是不可逆的状态 */
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;

        /**
         * Node状态字段，包括如下值:
         *   SIGNAL:     该节点的后继节点是（或即将是）阻塞状态（通过LockSupport.park方法）, 
         *               当前节点释放锁或者被取消时，需要调用unpark函数来激活它的后继节点。
                         为了避免锁竞争，acquire方法必须明确指出一个信号。
         *   CANCELLED:  节点因为中断或者超时变更为取消状态.
         *               canceled是最终态，不可逆变为其他状态. 被取消的节点不会继续阻塞.
         *   CONDITION:  节点当前在条件队列中（condition queue）.这个节点不会被当做同步队列节点
         								 直到它从条件队列中释放，这时waitStatus的值被设置为0.
         *               It will not be used as a sync queue node
         *               until transferred, at which time the status
         *               will be set to 0.
         *   PROPAGATE:  releaseShared应该传播到其他节点。 
         								 在doReleaseShared中对此进行了设置（仅适用于头节点），
         								 以确保传播继续进行，即使此后进行了其他操作也是如此。
         *   0:          除以上四种情况外
         */
        volatile int waitStatus;
    
        volatile Node prev;
        volatile Node next;

        /**
         * 等待锁的线程
         */
        volatile Thread thread;

        /**
     		 * condition的阻塞单向链表，阻塞队列只有在独占模式下才有（AQS还有共享模式）
         */
        Node nextWaiter;
    }

}

Суммировать

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

Цитировать

CLH, блокировка очереди MCS Введение

Спин-блокировки (блокировки CLH и блокировки MCS) в архитектуре UMA и архитектуре NUMA

Платформа синхронизации JUC

Документ java.util.concurrent Synchronizer Framework

Doug Lea's Home Page

Разница между Thread.sleep, Object.wait, LockSupport.park

Принцип и реализация блокировки MCS

Оптимистический замок и пессимистический замок необходимы для интервью

Вы действительно понимаете ReentrantReadWriteLock?

Блокировка чтения-записи Java Анализ исходного кода ReentrantReadWriteLock

Категории