zyxist.com

LMAX Disruptor to implementacja kolejki zdarzeń dla języka Java, której pojawienie się wywołało w Internecie dużo szumu z uwagi na innowacyjne podejście do tematu współbieżności. Biblioteka została zaprojektowana przez firmę LMAX Exchange Inc. na potrzeby ich flagowego produktu, platformy do handlu instrumentami finansowymi. Zespół pracujący nad nią musiał rozwiązać poważny problem, mianowicie w jaki sposób obsłużyć do 6 milionów operacji handlowych na sekundę, z czego każda z nich zmienia warunki rynkowe dla pozostałych. Architektura, której użyli, bazuje na tzw. event sourcingu - cała logika biznesowa przetwarzana jest przez jeden wątek (!), który trzyma wszystkie dane w pamięci. Disruptor odgrywa tutaj kluczową rolę dostarczania mu informacji o kolejnych zdarzeniach oraz wypychania odpowiedzi na świat. Kolejka zdarzeń została zrealizowana jako preinicjalizowany bufor cykliczny, zoptymalizowany pod kątem efektywnego cache'owania przez procesory oraz niewykorzystywanie zamków. Nie będę tutaj szczegółowo wgryzał się w teorię, gdyż to zagadnienie zostało świetnie opisane przez Martina Fowlera w jego artykule.

Gdy eksperymentowałem z biblioteką, chwilę czasu zajęło mi rozeznanie, jak połączyć trzech konsumentów czytających zdarzenia z bufora w taki sposób, by jeden z nich nigdy nie wyprzedził dwóch pozostałych podczas czytania. Innymi słowy, chciałem, aby jeden z konsumentów mógł odebrać dane zdarzenie tylko wtedy, gdy skończyli je przetwarzać pozostali. Wiedziałem, że biblioteka umożliwia to, ponieważ jest to jedna z podstawowych funkcjonalności potrzebnych LMAX-owi do zrealizowania ich systemu. Niestety, API biblioteki trochę się pozmieniało przez kilka lat, a wszystkie blogi oraz tutoriale, do jakich dotarłem, opisywały stare podejście. Do olśnienia wystarczyło wpisanie kropki w IDE - moim oczom ukazała się lista dostępnych metod, wśród których znajdowało się kilka ciekawych metod tworzących prosty DSL do komponowania barier. Poniżej znajdziesz prosty przykład:

public class DisruptorTest {
   public static void main(String[] args) {
      int bufferSize = 1024;      
      Disruptor<Event> disruptor = new Disruptor<>(
         Event::new, bufferSize, Executors.defaultThreadFactory()
      );

      // 1
      disruptor
          .handleEventsWith(new Consumer("A"), new Consumer("B"))
          .then(new Consumer("C"));

      disruptor.start();
      produceSomeEvents(disruptor);
      disruptor.shutdown();
   }

   private static void produceSomeEvents(Disruptor<Event> disruptor) {
      // 2
      RingBuffer<Event> ringBuffer = disruptor.getRingBuffer();
      Producer producer = new Producer();
      for (int i = 0; i < 1000; i++) {
         ringBuffer.publishEvent(producer);
      }
   }

   public static class Event {
      private int value;
   }

   public static class Producer implements EventTranslator<Event> {
      private int i = 0;

      @Override
      public void translateTo(Event event, long sequence) {
         event.value = i++;
      }
   }

   public static class Consumer implements EventHandler<Event> {
      private final String name;

      public Consumer(String name) {
         this.name = name;
      }

      @Override
      public void onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
         System.out.println(name+": " + event.value);
      }
   }
}

Wyjaśnienie:

1. Tutaj komponujemy ustawienie naszych trzech konsumentów. A oraz B są niezależne od siebie, dlatego mogą się dowolnie wyprzedzać podczas czytania pierścienia, podczas gdy C zawsze musi na nich zaczekać.
2. Publikuj w pierścieniu zdarzenia z liczbami od 0 do 1000.

Wszyscy konsumenci po prostu wypisują odczytany ze zdarzenia numer oraz swoją nazwę. Jeśli uruchomisz ten przykład, zobaczysz, że C zawsze wypisuje liczby, które zostały już wcześniej wypisane przez A oraz B, podczas gdy liczby wypisywane przez A i B mogą się dowolnie przeplatać - raz daną liczbę jako pierwszy wypisze A, innym razem B. Wszystko jest realizowane dzięki odrobinie magii z łańcuchem metod handleEventsWith() oraz then().

Bariery bardzo przydają się do budowania niezawodnych systemów przetwarzania zdarzeń. Jeden z konsumentów może zapisywać zdarzenia do dziennika, drugi zaś je obsługiwać. Do dalszej obsługi mogą trafić tylko te zdarzenia, które zostały już zapisane w dzienniku tak, aby w przypadku awarii można je było odtworzyć. Bez tego ryzykowalibyśmy utratę części danych. Dlatego - używaj barier.