Я действительно нашел время, чтобы изучить настоящий источник, из чистого любопытства, и идея этого довольно проста. Самая последняя версия на момент написания этого поста: 3.2.1.
Существует буфер, в котором хранятся предварительно распределенные события, которые будут содержать данные для чтения потребителями.
Буфер поддерживается массивом флагов (целочисленным массивом) его длины, который описывает доступность слотов буфера (подробнее см. Далее). Доступ к массиву осуществляется как java # AtomicIntegerArray, поэтому для целей данного объяснения вы можете также предположить, что он равен единице.
Может быть любое количество производителей. Когда производитель хочет записать в буфер, генерируется длинное число (как при вызове AtomicLong # getAndIncrement, Disruptor фактически использует свою собственную реализацию, но работает аналогичным образом). Давайте назовем это сгенерированное длинным как продюсером. Аналогичным образом, customerCallId генерируется, когда потребитель заканчивает чтение слота из буфера. Доступ к последнему customerCallId.
(Если есть много потребителей, выбирается звонок с самым низким идентификатором.)
Затем эти идентификаторы сравниваются, и, если разница между ними меньше, чем на стороне буфера, производителю разрешается писать.
(Если providerCallId больше, чем недавний customerCallId + bufferSize, это означает, что буфер заполнен, и производитель вынужден ждать шины, пока не станет доступным место.)
Затем производителю назначается слот в буфере на основе его callId (который является prducerCallId по модулю bufferSize, но, поскольку bufferSize всегда имеет степень 2 (ограничение принудительно устанавливается при создании буфера), используемой операциейallall является ManufacturerCallId & ( bufferSize - 1)). После этого можно свободно изменять событие в этом слоте.
(Фактический алгоритм немного сложнее, включающий кэширование недавнего потребителя-идентификатора в отдельном атомарном справочнике для целей оптимизации.)
Когда событие было изменено, изменение «опубликовано». При публикации соответствующего слота в массиве флагов заполняется обновленный флаг. Значение флага - это номер цикла (providerCallId, деленный на bufferSize (опять же, поскольку bufferSize имеет степень 2, фактическая операция - сдвиг вправо).
Аналогичным образом может быть любое количество потребителей. Каждый раз, когда потребитель хочет получить доступ к буферу, генерируется customerCallId (в зависимости от того, как потребители были добавлены в разрушитель, атом, используемый в генерации идентификатора, может быть общим или отдельным для каждого из них). Затем этот customerCallId сравнивается с последним producentCallId, и, если он меньше двух, читателю разрешается прогрессировать.
(Точно так же, если providerCallId равен даже customerCallId, это означает, что буфер пуст, а потребитель вынужден ждать. Способ ожидания определяется WaitStrategy во время создания прерывателя.)
Для отдельных потребителей (тех, у кого есть собственный генератор идентификаторов), следующая проверенная вещь - это возможность пакетного потребления. Слоты в буфере проверяются по порядку, начиная с единицы, соответствующей customerCallId (индекс определяется таким же образом, как и для производителей), до той позиции, которая соответствует недавнему providerCallId.
Они проверяются в цикле путем сравнения значения флага, записанного в массиве флага, со значением флага, созданным для consumerCallId. Если флаги совпадают, это означает, что производители, заполняющие слоты, передали свои изменения. Если нет, цикл прерывается и возвращается самый высокий зафиксированный changeId. Слоты от ConsumerCallId до полученных в changeId можно использовать в пакетном режиме.
Если группа потребителей читает вместе (те, у кого общий генератор идентификаторов), то каждый из них принимает только один callId, и проверяется и возвращается только слот для этого единственного callId.