LMAX Disruptor 튜터리얼

LMAX Disruptor 튜터리얼

https://github.com/LMAX-Exchange/disruptor/wiki/Getting-Started 를 번역 하였다.

기본 이벤트 프로듀스(생산)와 컨슘(소비)

Disruptor에 대해 알아보기 위해 Disruptor를 이용해 가장 간단하고 테스트를 하기 쉬운 예제로 알아볼 예정입니다.

예제 중 하나는 1개의 long타입의 값을 프로듀서(producer)에서 컨슈머(consumer)로 보낼 것입니다.

컨슈머는 간단하게 값을 출력하기만 할 것입니다.

먼저 데이터를 들고갈 오브젝트를 만저 만들어 볼 것입니다.

LongEvent.java

1 2 3 4 5 6 7

import lombok.Data;   @Data public class LongEvent {     private long value; }

cs

Disruptor가 이 이벤트를 우리에게 할당 하려면 EventFactory를 만들어야 한다 EventFactory는 LongEvent를 만들어 줍니다.

LongEventFactory.java

1 2 3 4 5 6 7 8 9

import com.lmax.disruptor.EventFactory;   public class LongEventFactory implements EventFactory<LongEvent> {       @Override     public LongEvent newInstance() {         return new LongEvent();     } } Colored by Color Scripter

cs

Event가 만들어지면 만들어진 이벤트들을 처리할 Event 컨슈머(consumer)를 만들어야 한다. 이번 예제에서 필요한건 콘솔에 값을 출력하는 컨슈머입니다.

LongEventHandler.java

1 2 3 4 5 6 7 8 9

import com.lmax.disruptor.EventHandler;   public class LongEventHandler implements EventHandler<LongEvent> {       @Override     public void onEvent(LongEvent longEvent, long sequence, boolean endofBatch) throws Exception {         System.out.println("event:" + longEvent);     } } Colored by Color Scripter

cs

데이터는 주로 http request나 Queue에서 오지만 예제를 해보아야 하기 때문에 데이터를 주는 프로듀서(Producer)를 만들어야 합니다.

LongEventProducer.java

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

public class LongEventProducer {     private final RingBuffer<LongEvent> ringBuffer;         public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {         this.ringBuffer = ringBuffer;     }       public void onData(ByteBuffer bb) {         long sequence = ringBuffer.next();           LongEvent event = ringBuffer.get(sequence);         event.setValue(bb.getLong(0));           ringBuffer.publish(sequence);     } } Colored by Color Scripter

cs

위와 같은 코드를 써야 하는데 이렇게 사용하면 이벤트 발행이 단순 큐(queue)를 사용 하는 것 보다 더 복잡하고 의존성이 높아집니다.

이건 왜그러냐면 이벤트를 미리 할당 하기 때문입니다. 이벤트를 미리 할당 하려면 메세지 발행을 하는데 2개의 단계를 거쳐야 합니다. 예를 들어 링 버퍼에 슬롯을 만들고 데이터를 발행 해야 하는 과정이 필요 합니다.

그러면 데이터를 발행 하는 명령은 try/catch를 걸어주어야 합니다.

링버퍼에 슬롯을 만들면(RingBuffer.next()를 호출하면) 순서를 반드시 함께 넘겨 주어야 합니다.

이 과정대로 처리를 하지 않으면 디스럽터의 상태가 오염됩니다. 멀티 프로듀서 환경에서는 재시작 하지 않으면 복구가 되지 않는 문제가 발생할 수 있습니다.

버전3.0 트렌스레이터스(Translators) 사용하기

링버퍼의 위와 같은 문제를 해결하기 위해 버젼3.0 디스럽터에는 람다 스타일의 API가 추가 되었습니다. 이 API를 이용하면 개발자가 위와 같이 복잡한 로직을 캡슐링 해서 처리 할 수 있습니다.

3.0버전은 Event Publisher/ Event Translator API를 사용하는게 좋습니다.

import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
 
import java.nio.ByteBuffer;
 
public class LongEventProducerWithTranslator {
    private final RingBuffer<LongEvent> ringBuffer;
 
    public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }
 
    private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR =
            new EventTranslatorOneArg<LongEvent, ByteBuffer>() {
                @Override
                public void translateTo(LongEvent longEvent, long l, ByteBuffer byteBuffer) {
                    longEvent.setValue(byteBuffer.getLong(0));
                }
            };
 
    public void onData(ByteBuffer byteBuffer) {
        ringBuffer.publishEvent(TRANSLATOR, byteBuffer);
    }
}
 

람다 스타일 API를 사용하는 또 다른 장점은 트렌스레이터 코드를 분리된 클래스로 뽑아내서 비교적 쉽게 유닛 테스트를 만들 수 있다는 것입니다.

디스럽터는 EventTranslator, EventTranslatorOneArg, EventTranslatorTwoArg등 트렌스레이터를 만들 수 있는 여러가지 인터페이스를 제공합니다.

이게 가능한 이유는 트렌스레이터가 스테텍 클래스 또는 넌 캡춰링 람다(외부 변수에 액세스 불가)로 되어있기 때문에 트렌스레이터를 통해 링버퍼로 들어갈 수 있습니다.

마지막 단계는 위에 만들었던 클래스들을 한군데 모으는겁니다. 모든 컴포넌트를 수동으로 불러와도 되지만 이건 복잡하기 때문에 DSL이 간단한 생성자를 제공하는걸 사용합니다.

DSL을 이용하면 몇몇 복잡한 케이스는 불가능 하지만 대부분의 상황에서 사용할 수 있습니다.

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
 
import java.lang.reflect.Executable;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
 
public class LongEventMain {
    public static void main(String[] args) throws InterruptedException {
        Executor executor = Executors.newCachedThreadPool();
 
        LongEventFactory factory = new LongEventFactory();
 
        int bufferSize = 1024;
 
        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, executor);
 
        disruptor.handleEventsWith(new LongEventHandler());
 
        disruptor.start();
 
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
 
        LongEventProducer producer = new LongEventProducer(ringBuffer);
 
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; true; l++ ) {
            bb.putLong(0, l);
            producer.onData(bb);
            Thread.sleep(1000);
        }
    }
}

end.