Оригинальная статья и краткое изложение опыта и жизненные перипетии от набора в школу до фабрики А

Нажмите, чтобы узнать подробностиwww.codercc.com

1. Введение в ConcurrentLinkedQueue

В однопоточном программировании мы часто используем некоторые классы коллекций, такие как ArrayList, HashMap и т. д., но эти классы не являются потокобезопасными. В интервью часто встречаются контрольные точки, например, ArrayList не является потокобезопасным, а Vector — потокобезопасным. Способ обеспечить потокобезопасность Vector — использовать синхронизированную эксклюзивную блокировку метода, чтобы превратить многопоточное выполнение в сериализацию. Чтобы сделать ArrayList потокобезопасным, вы также можете использоватьCollections.synchronizedList(List<T> list)Метод ArrayList преобразуется в потокобезопасный, но этот метод преобразования по-прежнему реализуется через метод синхронизированной модификации.Очевидно, что это неэффективный метод.В то же время очередь также является часто используемой структурой данных.Для решить проблему потокобезопасности, Мастер Дуг Ли подготовил для нас потокобезопасную очередь ConcurrentLinkedQueue. Из названия класса видно, что структура данных, реализующая очередь, имеет цепочку.

1.1 Node

Если вы хотите сначала изучить ConcurrentLinkedQueue, вы должны сначала взглянуть на его класс узла и понять его базовую структуру данных. Исходный код класса Node:

private static class Node<E> {
        volatile E item;
        volatile Node<E> next;
		.......
}

Узел Node в основном содержит два поля: одно — элемент поля данных, а другое — указатель next, который используется для указания на следующий узел для формирования цепной очереди. И все они украшены volatile для обеспечения видимости памяти (о volatileВы можете увидеть эту статью). Кроме того, ConcurrentLinkedQueue содержит следующие две переменные-члены:

private transient volatile Node<E> head;
private transient volatile Node<E> tail;

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

public ConcurrentLinkedQueue() {
    head = tail = new Node<E>(null);
}

Указатели начала и конца будут указывать на узел, поле элемента которого равно null.В настоящее время состояние ConcurrentLinkedQueue показано на следующем рисунке:

Как показано на рисунке, начало и конец указывают на один и тот же узел Node0, поле item этого узла равно null, а следующее поле равно null.

1.ConcurrentLinkedQueue初始化状态.png — 1. Состояние инициализации ConcurrentLinkedQueue.png

1.2 Несколько операций CAS, которые управляют узлом

Когда очередь удаляется из очереди и ставится в очередь, неизбежно, что узел должен работать, и проблема безопасности потоков легко возникает в многопоточности. Видно, что набор инструкций процессора может поддерживатьCMPXCHGПосле инструкции операция CAS используется, когда в исходном коде java задействована параллельная обработка.(См. Раздел 3.1 этой статьи для операций CAS), то есть несколько операций CAS на узле в CONCULRENTLINKEDQUEUE:

//更改Node中的数据域item
boolean casItem(E cmp, E val) {
    return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}
//更改Node中的指针域next
void lazySetNext(Node<E> val) {
    UNSAFE.putOrderedObject(this, nextOffset, val);
}
//更改Node中的指针域next
boolean casNext(Node<E> cmp, Node<E> val) {
    return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}

Можно увидеть, что эти методы на самом деле вызываются методами экземпляра UNSAFE, UNSAFE — этоsun.misc.Unsafeкласс, этот класс является базовым методом точки доступа, насколько вы можете понять его.Полезно знать, что работа CAS обеспечивается этим классом в конечном счете.

2. метод предложения

Для очереди вставка удовлетворяет свойству FIFO, вставка элементов всегда вставляется в конец очереди, а извлечение (удаление) элементов всегда происходит из головы очереди. Все, чтобы полностью понять ConcurrentLinkedQueue, естественно, начинается с метода предложения и метода опроса. Затем, чтобы понять метод предложения, используйте метод отладки для чтения кода построчно. Кроме того, при просмотре многопоточного кода вы можете использовать такой способ мышления:

однопоточное предложение > Многопоточное предложение > Некоторые темы предлагают, некоторые темы проводят опрос----предложение быстрее, чем опрос --------Длина очереди будет становиться все длиннее и длиннее, потому что узел предложения всегда находится в конце очереди, а узел опроса всегда в начале очереди, а это означает, что " пересечение между потоком предложения и потоком опроса. " ", то есть два типа потоков не влияют друг на друга. С точки зрения относительной скорости эта ситуация является "предложением одного потока" ----предложение медленнее, чем опрос Относительная скорость --------опроса выше, чем у предложения, то есть скорость удаления головы очереди выше скорости добавления узлов в конец очереди. что длина очереди будет становиться все короче и короче, а поток оффера и опрос Поток будет казаться «пересечением», то есть в этот момент его можно назвать узлом, в котором поток оффера и поток опроса работают одновременно .критическая точка, и поток предложения и поток опроса должны влиять друг на друга в этом узле. В соответствии с относительным порядком предложения и опроса в критической точке мы можем думать с двух точек зрения:1. Порядок выполнения: предложение-->опрос-->предложение, то есть, когда поток предложения вставляет Node2 после Node1, поток опроса в это время удалил Node1, что, очевидно, нужно учитывать в методе предложения;2. Порядок выполнения может быть: опрос-->предложение-->опрос, то есть когда поток опроса готов удалить узел (очередь является пустой очередью), в это время поток предложения вставляет узел, чтобы сделать очередь непустой.

Сначала посмотрите на этот фрагмент кода:

1. ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
2. queue.offer(1);
3. queue.offer(2);

Создайте экземпляр ConcurrentLinkedQueue, сначала предложите 1, затем предложите 2. Исходный код предложения:

public boolean offer(E e) {
1.    checkNotNull(e);
2.    final Node<E> newNode = new Node<E>(e);
3.    for (Node<E> t = tail, p = t;;) {
4.        Node<E> q = p.next;
5.        if (q == null) {
6.            // p is last node
7.            if (p.casNext(null, newNode)) {
                // Successful CAS is the linearization point
                // for e to become an element of this queue,
               // and for newNode to become "live".
8.                if (p != t) // hop two nodes at a time
9.                    casTail(t, newNode);  // Failure is OK.
10.                return true;
            }
            // Lost CAS race to another thread; re-read next
        }
11.        else if (p == q)
            // We have fallen off list.  If tail is unchanged, it
            // will also be off-list, in which case we need to
            // jump to head, from which all live nodes are always
            // reachable.  Else the new tail is a better bet.
12.            p = (t != (t = tail)) ? t : head;
           else
            // Check for tail updates after two hops.
13.            p = (p != t && t != (t = tail)) ? t : q;
    }
}

Анализ перспективы выполнения одного потока:

начать сПерспектива однопоточного выполненияСмотрите, процесс анализа предложения 1. Первая строка кода будет судить, является ли оно нулевым. Если оно равно нулю, будет напрямую выдано исключение нулевого указателя. Вторая строка кода поместит e в класс Node, а третья строка кода будет циклом for. . Есть только условия инициализации и нет условий завершения цикла. Это соответствует «рутине» CAS. Если операция CAS успешна в теле цикла, она вернется напрямую. Если операция CAS не удалась, она продолжит повторите попытку в цикле for, пока не добьетесь успеха. Здесь переменная экземпляра t инициализируется значением tail, а p инициализируется значением t, которое является хвостом. Чтобы облегчить следующее понимание,p считается реальным хвостовым узлом очереди, хвост не обязательно указывает на реальный хвостовой узел объекта, потому что хвост обновляется с задержкой в ConcurrentLinkedQueue, давайте посмотрим на конкретные причины. Когда код переходит к строке 3, и t, и p указывают на Node0, чье поле item имеет значение null, а поле next равно null, соответственно, созданное во время инициализации. Переменной 4-й строки q присваивается значение null, 5-я строка считается истинной, а 7-я строка использует casNext для установки вставленного узла в качестве следующего узла хвостового узла текущей очереди p. цикл заканчивается в следующий раз, когда Retry в цикле. Операция CAS успешно переходит к строке 8. В это время p==t, если суждение ложно, и сразу возвращает true. Если 1 вставлено успешно, состояние ConcurrentLinkedQueue в это время показано на следующем рисунке:

2.offer 1后队列的状态.png — 2. Статус очереди после оффера 1.png

Как показано на рисунке, хвостовой узел очереди в это время должен быть Node1, а узел, на который указывает tail, по-прежнему Node0, поэтому можно показать, что хвост обновляется с задержкой. Затем мы продолжаем рассматривать ситуацию предложения 2. Очевидно, что узел, на который указывает q в строке 4, не является нулевым, а указывает на узел Node1, если суждение в строке 5 ложно, если суждение в строке 11 ложно, код перейдет к строке 13. хорошо,При вставке узла мы зададим себе такой вопрос? Выше было объяснено, что хвост не указывает на реальный хвостовой узел очереди, поэтому при вставке узла мы должны сначала найти, куда можно вставить текущий хвостовой узел очереди?Затем строка 13 кодаУзнать настоящий хвостовой узел очереди.

Найдите реальный противоположный хвостовой узел очереди

p = (p != t && t != (t = tail)) ? t : q;

Давайте проанализируем эту строку кода, если этот код находится воднопоточная средаПри выполнении очевидно, что, поскольку p==t, p будет присвоено q в это время, а q равноNode<E> q = p.next, то есть Node1. В первом цикле указатель p указывает на Node1, реальный хвостовой узел очереди, затем в следующем цикле узел, на который указывает q в строке 4, равен нулю, затем в строке 5, если суждение верно, то в строке 7 он по-прежнему устанавливает следующий узел p для текущего добавленного узла с помощью метода casNext, а затем переходит к строке 8. В это время p!=t, 8-я строка, если она считается истинной, пройдетcasTail(t, newNode)Установите текущий узел Node в качестве хвостового узла очереди, и схематическая диаграмма состояния очереди в это время показана на следующем рисунке: 3.队列offer 2后的状态.png

Узел, на который указывает хвост, изменяется с Node0 на Node2., Причина, по которой сбой casTail здесь не нужно повторять, заключается в том, что код предложения в основном проходит через следующий узел q(Node<E> q = p.next) определяет последующее логическое направление.При отказе casTail диаграмма состояний выглядит следующим образом: 4.队列进行入队操作后casTail失败后的状态图.png

Как показано на рисунке,Если casTail не может установить здесь хвост, то есть хвост по-прежнему указывает на узел Node0, это не что иное, как несколько циклов, чтобы найти хвостовой узел через 13 строк кода..

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

Если следующий узел (следующее поле) узла, на который указывает хвост, равен нулю, это означает, что узел, на который указывает хвост, является реальным хвостовым узлом очереди, поэтому вставляемый узел можно вставить через casNext, но хвост в это время не меняется, как показано на рисунке 2;
Если следующий узел (следующее поле) узла, на который указывает хвост, не равен нулю, это означает, что узел, на который указывает хвост, не является реальным хвостовым узлом очереди. пройти черезq（Node<E> q = p.next）Передайте указатель вперед, чтобы найти хвостовой узел очереди, затем вставьте узел, который в настоящее время должен быть вставлен, через casNext и измените хвост через casTail, как показано на рисунке 3..

давайте оглянемся назадp = (p != t && t != (t = tail)) ? t : q;Эта строка кода находится в одном потоке, этот код никогда не назначит p на t, поэтому запись этого не будет иметь никакого эффекта, тогда мы попытаемсяМногопоточностьанализ в случае.

Анализ с точки зрения многопоточного выполнения

Многопоточное предложение

Очевидно, в этом сочинении есть более глубокий смысл, на самом деле,многопоточная средаСледующая строка кода интересна.t != (t = tail)эта операцияне атомарная операция, есть такая ситуация:

5.线程A和线程B有可能的执行时序.png — 5. Возможное время выполнения потока A и потока B.png

Как показано на рисунке, если предположить, что поток A читает переменную t в это время, а поток B просто предлагает Node в это время, он изменит хвостовой указатель в это время, а затем в это время, когда поток A выполнит t= tail снова, t будет указывать на другой узел Очевидно, что узлы, на которые указывает переменная t, прочитанная дважды до и после потока A, различны, т. е.t != (t = tail)верно, а так как t указывает на смену узлаp != tЭто также верно.В настоящее время результатом выполнения этой строки кода является то, что последние t указателей p и t указывают на один и тот же узел, и t также является реальным хвостовым узлом очереди в это время. Итак, теперь, когда обнаружен реальный хвостовой узел очереди, можно выполнить операцию предложения.

offer->poll->offer

Далее остается еще не проанализированная нами 11 строка кода, и можно примерно догадаться, что она должна быть ответомЧасть предложения темы, часть опросаданной ситуации. когдаif (p == q)Когда это правда, это означает, что следующий узел, на который указывает p, также указывает на себя.Такой тип узла называетсяДозорный узел,Этот тип узла имеет небольшую ценность в очереди и обычно выражается как узел, подлежащий удалению, или как пустой узел.. Чтобы хорошо разобраться в этой ситуации, давайте сначала посмотрим на процесс выполнения метода опроса, а потом оглянемся назад, короче, это очень интересная штука :).

3. метод опроса

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

public E poll() {
    restartFromHead:
    1. for (;;) {
    2.    for (Node<E> h = head, p = h, q;;) {
    3.        E item = p.item;
4.        if (item != null &amp;&amp; p.casItem(item, null)) {
            // Successful CAS is the linearization point
            // for item to be removed from this queue.
5.            if (p != h) // hop two nodes at a time
6.                updateHead(h, ((q = p.next) != null) ? q : p);
7.            return item;
        }
8.        else if ((q = p.next) == null) {
9.            updateHead(h, p);
10.            return null;
        }
11.        else if (p == q)
12.            continue restartFromHead;
        else
13.            p = q;
    }
}

скопировать код}

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

6.队列初始状态.png — 6. Исходное состояние очереди.png

Определение параметра предлагаем, мы все еще первыеПеременная p используется в качестве очереди для удаления реального головного узла очереди.Узел, на который указывает h (head), не обязательно является головным узлом очереди.. Давайте сначала рассмотрим ситуацию, когда опрашивается Node1, потому чтоp=h=head, ссылаясь на приведенный выше рисунок, очевидно, что поле данных Node1, на которое указывает p, в это время не равно нулю, в 4-й строке кодаitem!=nullПосле того, как он признан верным, следующий проходcasItemУстановите для поля данных Node1 значение null. Если настройка CAS дает сбой, цикл завершается и ждет повторения следующего цикла. Если выполнение строки 4 выполнено успешно, он вводит код строки 5. В это время как p, так и h указывают на Node 1. Решение if в строке 5 ложно, а затем напрямую возвращается к полю данных 1 Node1 в строка 7. Метод завершен, статус очереди показан ниже.

7.队列出队操作后的状态.png — 7. Состояние после удаления очереди.png

Продолжить опрос из очереди.Очевидно, что поле данных Node1, на которое указывают h и p, равно null, поэтому первое, что нужно сделать, этоНайдите головной узел, который нужно удалить (найдите узел, поле данных которого не равно нулю).

Найдите удаленный головной узел

Продолжайте смотреть, третья строка кода имеет значение null, четвертая строка кода, если она признана ложной, переходите к восьмой строке кода (q = p.next) if также ложно. Поскольку q указывает на Node2, суждение if в строке 11 также ложно, поэтому код переходит к строке 13. В это время p и q вместе указывают на Node2, а реальный узел, который нужно удалить, найден головной узел. Можно сделать вывод, что процесс поиска удаляемого головного узла выглядит следующим образом:Если поле данных текущего узла равно нулю, очевидно, что узел не является удаляемым узлом, и для проверки используется следующий узел текущего узла.. После первого цикла диаграмма состояний выглядит следующим образом:

8.经过一次循环后的状态.png — 8. Статус после одного цикла.png

Выполните следующий цикл, операция строки 4 такая же, как и выше, в настоящее время предполагается, что установка casItem в строке 4 выполнена успешно, поскольку p указывает на Node2, а h все еще указывает на Node1, в это время if в строке 5 считается верным, а затем выполнитьupdateHead(h, ((q = p.next) != null) ? q : p), Node3, на который в данный момент указывает q, в метод updateHead передается только ссылка h, указывающая на Node1, и ссылка q, указывающая на Node3. Исходный код метода updateHead:

final void updateHead(Node<E> h, Node<E> p) {
    if (h != p && casHead(h, p))
        h.lazySetNext(h);
}

Этот метод осуществляется в основном за счетcasHeadУкажите голову очереди на Node3 и передайтеh.lazySetNextУкажите следующее поле Node1 на себя. Наконец, верните значение Node2 в строке 7 кода. Состояние очереди в это время показано на следующем рисунке:

9.Node2从队列中出队后的状态.png — 9. Состояние после исключения Node2 из очереди.png

Следующее поле Node1 указывает на себя, а head указывает на Node3. Если очередь пуста, она будет выполняться до строки 8 кода.(q = p.next) == null, если считается истинным, поэтому null возвращается непосредственно в строке 10. Вышеприведенный анализ с точки зрения однопоточного выполнения, а также позволяет нам понять общую идею опроса.Теперь давайте подведем итоги:

Если Item узла, на который указывают head, h и p, не равен нулю, это означает, что узел является настоящим головным узлом (удаляемым узлом), просто установите для поля элемента значение null с помощью метода casItem и затем установите исходный элемент. Просто вернитесь назад.
Если элемент узла, на который указывают head, h и p, равен нулю, это означает, что узел не является реальным узлом, подлежащим удалению, поэтому необходимо найти узел, элемент которого не равен нулю. Протестируйте, позволив q указывать на следующий узел p (q = p.next), и если он найден, обновите узел, на который указывает head, с помощью метода updateHead и создайте дозорный узел (通过updateHead方法的h.lazySetNext(h)).

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

Анализ многопоточного выполнения:

Опрос в несколько тем

Теперь, оглядываясь назад на исходный код метода опроса, есть эта часть:

else if (p == q)
    continue restartFromHead;

Эта часть имеет дело с многопоточным опросом,q = p.nextДругими словами, q всегда указывает на следующий узел p, так при каких обстоятельствах p и q будут указывать на один и тот же узел? Согласно нашему анализу выше, только узел, на который указывает p, превращается в опрос.Дозорный узел(через h.lazySetNext в методе updateHead). Когда нить А судитp==qКогда поток B завершит выполнение метода опроса для преобразования узла, на который указывает p, вДозорный узелИ узел, на который указывает заголовок, изменился, поэтому его необходимо выполнить из restartFromHead, чтобы убедиться, что используется последний заголовок.

poll->offer->poll

Представьте, есть такая ситуация, если текущая очередь пуста, поток А выполняет операцию опроса, пока поток Б выполняет предложение, а затем поток А выполняет опрос, то в это время поток А возвращает ноль или поток Б только что вставил Что о новейшем узле? Давайте напишем демо генерации:

public static void main(String[] args) {
    Thread thread1 = new Thread(() -> {
        Integer value = queue.poll();
        System.out.println(Thread.currentThread().getName() + " poll 的值为：" + value);
        System.out.println("queue当前是否为空队列：" + queue.isEmpty());
    });
    thread1.start();
    Thread thread2 = new Thread(() -> {
        queue.offer(1);
    });
    thread2.start();
}

Результат:

Значение опроса Thread-0: null Является ли очередь в настоящее время пустой очередью: false

Управляйте порядком выполнения потока thread1 и потока thread2 с помощью отладки, поток thread1 сначала выполняется до 8-й строки кода.if ((q = p.next) == null), так как очередь в это время пуста, если оценивается как истина, введите блок if.В это время сначала приостанавливается поток 1, а затем поток 2 выполняет предложение вставить узел со значением 1, и выполнение потока 2 заканчивается . Пусть thread1 снова выполнится, затемthread1 не пытался повторить, но код продолжает опускаться, возвращая null, хотя в это время очередь вставила новый узел со значением 1 из-за thread2. Таким образом, выходной результат опроса thread0 равен нулю, но очередь не является пустой очередью. следовательно,При определении того, пуста ли очередь, об этом нельзя судить по тому, что поток возвращает null при опросе, это можно определить с помощью метода isEmpty..

4. Некоторые потоки предлагают опрос некоторых потоков в методе offer

При анализе метода offer у нас осталась проблема – понимание 11-й строки кода в методе offer.

offer->poll->offer

Строка 11 метода предложенияif (p == q), который может сделать, если он признан истинным. Узел, на который указывает p, являетсяДозорный узел, и когда будет построен дозорный узел? При обсуждении метода опроса мы нашли ответ, то есть, когда поле элемента узла, на которое указывает заголовок, равно нулю, будет найден настоящий головной узел. будет обновлен, и заголовок будет обновлен.Узел, на который указывает исходный заголовок, устанавливается в качестве контрольного узла. ** Предположим, что начальное состояние очереди показано на следующем рисунке: 10.offer和poll相互影响分析时队列初始状态.png Следовательно, когда поток A выполняет предложение, поток B выполняет опрос, и возникает одна из следующих ситуаций: 11.线程A和线程B可能存在的执行时序.png

Как показано на рисунке, хвостовой узел потока A имеет следующий узел Node1, поэтому он будет искать настоящий хвостовой узел очереди, обращаясь к q. Когда выполнение достигает решенияif (p == q)В это время поток B выполняет операцию опроса. Для потока B head и p указывают на Node0. Поскольку поле элемента Node0 равно null, он также будет продвигаться вперед, чтобы найти реальный головной узел Node1 очереди и выполнить его. в потоке B. После опроса Node0 будет преобразован вДозорный узел, что означает, что заголовок очереди изменился, а статус очереди такой, как показано ниже.

12.线程B进行poll后队列的状态图.png — 12. Диаграмма состояний очереди после потока B polls.png

В это время поток A выполняет суждениеif (p == q)Когда это правда, он будет продолжать выполнятьсяp = (t != (t = tail)) ? t : head;, так как хвостовой указатель не изменился, p присваивается голове, и операция вставки снова начинается с головы.

5. Дизайн хмеля

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

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

** Время триггера обновления заголовка: ** Когда поле элемента узла, на которое указывает заголовок, равно нулю, будет выполнена операция обнаружения реального головного узла очереди.После того, как головной узел найден и удаление завершено, обновление заголовка будет выполняться через updateHead; Когда поле элемента узла, на которое указывает заголовок, не равно нулю, только удалите узел без обновления заголовка.

А во время операции обновления в исходном коде будут комментарии следующего вида:hop two nodes at a time. Итак, причина, по которой эта стратегия отложенного обновления называется HOPS, заключается в следующем (догадайтесь:)). интервал в середине получил один. Итак, какова цель этого дизайна?

Если хвост всегда используется в качестве хвостового узла очереди, объем кода для реализации будет меньше, а логика будет легче для понимания. Однако у этого есть недостаток.**Если выполняется большое количество операций постановки в очередь, каждый раз выполняется CAS для обновления хвоста, что также приведет к большой потере производительности. Если операцию обновления CAS можно сократить, эффективность операции входа в очередь может быть значительно повышена, поэтому мастер Дуг Ли использует CAS только для обновления хвоста каждый интервал (расстояние между хвостом и хвостовым узлом равно 1). **То же самое относится и к обновлению головы.Хотя в этом дизайне хвостовой узел очереди находится больше в цикле, общая эффективность операции чтения намного выше, чем производительность записи.Поэтому лишнее находится в цикле. снижение производительности при размещении хвостового узла посередине относительно невелико.

использованная литература

Искусство параллельного программирования на Java

«Программирование с высокой степенью параллелизма на Java»

Сообщение в блоге ConcurrentLinkedQueue: https://www.cnblogs.com/sunshine-2015/p/6067709.html