Запись с помощью одного потока LMAX

Я познакомился с LMAX и этой замечательной концепцией под названием RingBuffer. Итак, ребята, скажите, что при записи в кольцевой буфер только с одним потоком производительность намного лучше, чем с несколькими производителями...

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

Вопрос 1. Я мог что-то упустить или неправильно понять какой-то аспект, но если у вас есть N параллельных производителей, как можно объединить их в 1 и не блокировать друг друга?

Вопрос 2. Я вспоминаю observables rxJava, где можно было взять N observables и объединить их в 1, используя Observable.merge интересно, блокирует ли он или поддерживает какую-либо блокировку каким-либо образом?


java multithreading rx-java disruptor-pattern lmax
person vach    schedule 14.05.2015    source источник

Ответы (2)


arrow_upward
3
arrow_downward

Влияние на RingBuffer многопоточной записи незначительно, но при очень больших нагрузках может быть значительным.

Реализация RingBuffer содержит узел next, где будет сделано следующее добавление. Если только один поток записывает в кольцо, процесс всегда завершается за минимальное время, то есть buffer[head++] = newData.

Чтобы справиться с многопоточностью и избежать блокировок, вы обычно делаете что-то вроде while ( !buffer[head++].compareAndSet(null,newValue)){}. Этот замкнутый цикл продолжал выполняться, пока другие потоки мешали сохранению данных, что замедляло пропускную способность.

Обратите внимание, что выше я использовал псевдокод, взгляните на getFree в моей реализации здесь для реального примера.

  // Find the next free element and mark it not free.
  private Node<T> getFree() {
    Node<T> freeNode = head.get();
    int skipped = 0;
    // Stop when we hit the end of the list
    // ... or we successfully transit a node from free to not-free.
    // This is the loop that could cause delays under hight thread activity.
    while (skipped < capacity && !freeNode.free.compareAndSet(true, false)) {
      skipped += 1;
      freeNode = freeNode.next;
    }
    // ...
  }
person OldCurmudgeon    schedule 14.05.2015
comment
Таким образом, использование этого для N производителей (где n не очень велико) будет работать быстрее, чем блокировка доступа для записи? - person vach; 15.05.2015
comment
@vach - Да. Любой неблокирующий алгоритм предпочтительнее алгоритма, использующего блокировки. - person OldCurmudgeon; 15.05.2015
comment
Чего я не могу понять, так это того, как LMAX удается сделать это в одном потоке? они получают много заказов одновременно, как они получают их для кольцевого буфера в одном потоке? Есть мысли по этому поводу :)? - person vach; 19.05.2015
comment
@vach — здесь есть отличная статья. Это просто бизнес-логика, которая является однопоточной. - person OldCurmudgeon; 19.05.2015
comment
Прочитал 2 раза :) Просто сложилось впечатление, что каким-то образом удалось сделать производителя одним потоком (потому что они его и разработали) - person vach; 19.05.2015

arrow_upward
2
arrow_downward

Внутри RxJava слияние использует конструкцию сериализации, которую я называю emitter-loop. который использует synchronized и блокирует.

Наши «клиенты» используют слияние в основном в случаях, не чувствительных к пропускной способности и задержке, или полностью однопоточных, и блокировка на самом деле не является проблемой.

Можно написать неблокирующий сериализатор, который я называю queue-drain, но слияние нельзя настроить для использования вместо этого.

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

person akarnokd    schedule 14.05.2015