[Java] Reactor 패턴과 이벤트 루프

Reactor 패턴

Reactor 패턴은 동시에 들어오는 여러 종류의 이벤트를 처리하기 위한 동시성을 다루는 디자인 패턴 중 하나입니다.

Reactor 패턴은 관리하는 리소스에서 이벤트가 발생할 때까지 대기하다가 이벤트가 발생하면 해당 이벤트를 처리할 수 있는 핸들러(handler)에게 디스패치(dispatch)하는 방식으로 이벤트에 반응하는 패턴으로 '이벤트 팬들링(event handling)'패턴이라고도 부릅니다.

 

Reactor 패턴은 크게 Reactor와 핸들러로 구성됩니다.

 

• Reactor: 무한 반복문을 실행해 이벤트가 발생할 때까지 대기하다가 이벤트가 발생하면 처리할 수 있는 핸들러에게 디스패치합니다. 이벤트 루프라고도 부릅니다.

 

• 핸들러: 이벤트를 받아 필요한 비즈니스 로직을 수행합니다.
  세부적인 구현은 상황에 맞게 변경할 수 있습니다. 따라서 세부 구현 내용에 초점을 맞추기보다는 리소스에서 발생한 이벤트를 처리하기까지의 과정과, 그 과정에서 Reactor와 핸들러가 어떤 역할을 하는지 이해하는 것이 Reactor 패턴을 이해하는 데 더 많은 도움이 됩니다.

 

Reactor

Reactor.java

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Set;
  
public class Reactor implements Runnable {
    final Selector selector;
    final ServerSocketChannel serverSocketChannel;
  
    Reactor(int port) throws IOException {
        selector = Selector.open();
  
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        SelectionKey selectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  
        // Attach a handler to handle when an event occurs in ServerSocketChannel.
        selectionKey.attach(new AcceptHandler(selector, serverSocketChannel));
    }
  
    public void run() {
        try {
            while (true) {
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                for (SelectionKey selectionKey : selected) {
                    dispatch(selectionKey);
                }
                selected.clear();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
  
    void dispatch(SelectionKey selectionKey) {
        Handler handler = (Handler) selectionKey.attachment();
        handler.handle();
    }
}

• L19 ~ L26: 무한 반복문을 실행하며 Selector에서 이벤트가 발생하기까지 대기하다가 이벤트가 발생하는 경우 적절한 핸들러에서 처리할 수 있도록 dispatch합니다.
• L33 ~ L34: 'Attached object'로 등록돼 있던 핸들러를 가져와 비즈니스 로직을 처리합니다.

 

핸들러

이벤트를 받아 필요한 비즈니스 로직을 수행합니다. ServerSocketChannel에서 Accept 이벤트를 받는 경우 연결 요청을 처리할 핸들러를 구현하고, SocketChannel에서 데이터를 전달받는 경우 에코 응답을 처리할 핸들러를 구현합니다.

 

Handler.java

public interface Handler {
  
    void handle();
}

 

 

AcceptHandler.java

import java.io.IOException;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
  
class AcceptHandler implements Handler {
    final Selector selector;
    final ServerSocketChannel serverSocketChannel;
  
    AcceptHandler(Selector selector, ServerSocketChannel serverSocketChannel) {
        this.selector = selector;
        this.serverSocketChannel = serverSocketChannel;
    }
  
    @Override
    public void handle() {
        try {
            final SocketChannel socketChannel = serverSocketChannel.accept();
            if (socketChannel != null) {
                new EchoHandler(selector, socketChannel);
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

 

 

EchoHandler.java

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
  
public class EchoHandler implements Handler {
    static final int READING = 0, SENDING = 1;
  
    final SocketChannel socketChannel;
    final SelectionKey selectionKey;
    final ByteBuffer buffer = ByteBuffer.allocate(256);
    int state = READING;
  
    EchoHandler(Selector selector, SocketChannel socketChannel) throws IOException {
        this.socketChannel = socketChannel;
        this.socketChannel.configureBlocking(false);
        // Attach a handler to handle when an event occurs in SocketChannel.
        selectionKey = this.socketChannel.register(selector, SelectionKey.OP_READ);
        selectionKey.attach(this);
        selector.wakeup();
    }
  
    @Override
    public void handle() {
        try {
            if (state == READING) {
                read();
            } else if (state == SENDING) {
                send();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
  
    void read() throws IOException {
        int readCount = socketChannel.read(buffer);
        if (readCount > 0) {
            buffer.flip();
        }
        selectionKey.interestOps(SelectionKey.OP_WRITE);
        state = SENDING;
    }
  
    void send() throws IOException {
        socketChannel.write(buffer);
        buffer.clear();
        selectionKey.interestOps(SelectionKey.OP_READ);
        state = READING;
    }
}

 

이벤트 루프

멀티플렉싱 기반 처리로 하나의 스레드에서 다중 접속 서버를 구현하는 방법과 동시에 들어오는 여러 종류의 이벤트를 처리하기 위해 동시성을 다루는 디자인 패턴인 Reactor 패턴에 대해 알아보았습니다.

이전 글에서 만났던 이벤트 루프를 기억하시나요?

이벤트 루프
이벤트 루프(event loop)는 동시성(concurrency)를 제공하기 위한 프로그래밍 모델 중 하나로, 특정 이벤트가 발생할 때까지 대기하다가 이벤트가 발생하면 디스패치해 처리하는 방식으로 작동합니다.

 

어딘가 익숙한 설명이 맞죠? 맞습니다. 내부적으로 Selector를 이용해 특정 이벤트가 발생할 때까지 대기(block)하다가 이벤트가 발생하면 적절한 핸들러로 이벤트를 전달(dispatch)해 처리하는 역할을 무한 루프를 실행해 반복하던 Reactor가 바로 이벤트 루프 입니다.

 

 

다양한 이벤트 루프 구현체

여러 언어에 걸쳐 프레임워크나 라이브러리 형태로 제공되는 다양한 이벤트 루프 구현체가 존재합니다. 그중에서 대표적인 몇 가지 구현체를 소개합니다.

Netty

Netty

Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients. Netty is a NIO client server framework which enables quick and easy development of network applications such as protocol servers and clients. It greatly simplifies and streamlines network programming such as TCP and UDP socket server.

 

Netty는 비동기식 이벤트 기반 네트워크 애플리케이션 프레임워크입니다. Java를 사용하는 개발자라면 많이 접해 봤을 텐데요. Netty자체로도 많이 사용하지만 고성능 네트워크 처리를 위해 Armeria를 포함해서 정말 많은 프레임워크나 라이브러리에서 사용하고 있습니다. 이와 같은 Netty도 기본적으로 지금까지 살펴본 Java NIO의 Selector와 Reactor 패턴을 기반으로 구현돼 있습니다.

 

Node.js

Node.js

Node.js is a JavaScript runtime built on Chrome's V8 JavaScript engine.

 

JavaScript 런타임 환경을 제공하는 Node.js에도 이벤트 루프가 있습니다. 정확히는 Node.js에서 사용(참고)하는 libuv 라이브러리가 이벤트 루프를 제공합니다.

libuv

libuv is cross-platform support library which was originally written for Node.js. It’s designed around the event-driven asynchronous I/O model.

Feature highlights

• -Full-featured event loop backed by epoll, kqueue, IOCP, event ports.
• - ...

 

libuv는 비동기식 I/O를 멀티 플랫폼 환경에서 제공할 수 있도록 만든 라이브러리입니다. 내부적으로 멀티플렉싱을 위해 epoll, kqueue, IOCP(input/output completion port)를 사용하고 있습니다.

 

Redis

Redis

The open source, in-memory data store used by millions of developers as a database, cache, streaming engine, and message broker.

Redis는 이벤트 처리를 위해 자체적으로 이벤트 루프(참고)를 구현해서 사용하는 싱글 스레드 애플리케이션입니다.

typedef struct aeEventLoop
{
    int maxfd;
    long long timeEventNextId;
    aeFileEvent events[AE_SETSIZE]; /* Registered events */
    aeFiredEvent fired[AE_SETSIZE]; /* Fired events */
    aeTimeEvent *timeEventHead;
    int stop;
    void *apidata; /* This is used for polling API specific data */
    aeBeforeSleepProc *beforesleep;
} aeEventLoop;

 

 

이벤트 루프를 블록하면 안 되는 이유

이벤트 루프가 어떻게 구성되고 작동하는지 이해했기 때문에 이제 이벤트 루프 스레드를 블록한다는 것이 무엇을 의미하는지, 그리고 어떤 문제를 야기하는지 알 수 있습니다.

이벤트 루프는 다음 작업을 반복합니다.

1. 이벤트가 발생하기를 대기한다.
2. 이벤트가 발생하면 처리할 수 있는 핸들러에 이벤트를 디스패치한다.
3. 핸들러에서 이벤트를 처리한다.
4. 다시 1~3 단계를 반복한다.

1부에서 Armeria 기반으로 만든 애플리케이션에서 작성했던 예제를 위 이벤트 루프 작업 단계에 맞춰 살펴보면 다음과 같습니다.

1. Armeria 기반의 서버 애플리케이션이 실행돼 클라이언트의 요청이 들어오기를(네트워크 이벤트가 발생하기를) 기다린다.
2. 클라이언트의 요청으로 네트워크 I/O 이벤트가 발생하고 해당 이벤트를 처리할 수 있는 핸들러로 이벤트가 디스패치된다.
3. 핸들러에서 이벤트를 처리하는 과정에서 AuthenticationDecorator 로직이 실행된다.
   • 인증 로직을 수행하며 블로킹 I/O 방식으로 외부 API를 호출해서 응답이 오기 전까지 스레드를 블록한다.
   • 블록되는 스레드는 이벤트 루프를 실행하고 있는 스레드이기 때문에 블록이 해제될 때까지 1~3단계를 반복할 수 없게 된다.

결과적으로 핸들러에서 스레드 블록을 유발하는 작업이나 시간이 오래 걸리는 작업을 처리하는 경우에는 해당 시간 동안 이벤트 루프가 블록되기 때문에 발생한 이벤트를 처리하는 시간도 지연됩니다. 이벤트 루프가 블록되는 문제를 피하고 높은 응답성을 유지하기 위해서는 스레드 블록을 유발하는 작업은 이벤트 루프가 아닌 별도 스레드에서 수행해야 합니다. 또한 핸들러에 블록을 유발하는 작업은 없지만 이벤트를 처리하는 로직 자체가 CPU가 많이 필요한 작업을 포함하고 있어서 시간이 많이 필요한 경우에는 작업 범위를 분할해서 처리해야 합니다.