[Java] Multiplexer가 뭔가요?

I/O Multiplexing 이란?

 

I/O Multiplexing이라는 기술은 여러 개의 입출력 작업을 동시에 처리하는 기법을 의미합니다. 일반적으로 하나의 프로세스(혹은 싱글 스레드)가 여러 개의 I/O 요청을 처리해야 할 때 사용됩니다. 대표적인 예로는 Netty와 Redis와 같은 고성능 네트워크 애플리케이션이 있습니다.

 

그렇다면 Java에서는 I/O Multiplexing을 어떻게 구현할 수 있을까요?

 

Java에서는 I/O Multiplexing을 구현하기 위한 핵심 컴포넌트로 Java NIO의 Selector를 사용합니다. 또한 운영체제에 따라 Selector의 여러 가지 구현체를 제공합니다. 이 Selector를 사용하면 Java에서도 I/O Multiplexing이 가능한 EventLoop를 구현할 수 있습니다.

 

Java NIO의 Selector에 대해 알아보기 전에, I/O Multiplexing은 File Descriptor(fd)의 상태 변화를 비동기적으로 처리하는 방식으로 동작하며, 이를 위해 운영 체제의 system call이 사용되기 때문에 File Descriptor(fd) 와 system call에 대한 기본적인 이해가 필요합니다.

 

File Descriptor 란?

 

유닉스 계열의 OS에서는 일반적인 파일, 네트워크 소켓, 파이프, 블록 디바이스, 캐릭터 디바이스 등 모든 객체를 파일로 관리하는데 이러한 파일에 접근할 때 사용하는 추상적인 값을 fd(file descriptor)라고 합니다.

fd는 음이 아닌 정수를 사용하며 기본적으로 표준 입력(0), 표준 출력(1), 표준 에러(2)에 각각 0, 1, 2를 할당하기 때문에 프로세스가 파일을 open 하면 운영체제는 fd에 3을 시작으로 사용하지 않는 가장 작은 값을 할당해 줍니다.

 

public static void main(String[] args) {
    File file = new File(Main.class.getClassLoader().getResource("sample.txt").getFile());

    try (FileInputStream fis = new FileInputStream(file)) {
        FileDescriptor fd = fis.getFD();
        Field field = FileDescriptor.class.getDeclaredField("fd");
        field.setAccessible(true);
        int value = field.getInt(fd);
        log.info("fd: {}", value);
    } catch (Exception e){
    	log.error("error: {}", e.getMessage());
    }
}

 

Java에서는 FileInputStream의 getFd()로 FileDescriptor 객체에 접근이 가능하나, FileDescriptor 객체의 fd 필드는 접근제한자가 private으로 선언되어 있어 reflection을 사용하여 조회해야 합니다. 또한 Java 9부터는 모듈 시스템이 강화되었기 때문에 Java 8에서 테스트해보았습니다.

 

File Descriptor는 유닉스 계열의 OS(Linux, macOS)에서 사용되는 개념으로 윈도우 환경에서 테스트 시 제대로 동작하지 않아 기본값인 -1이 출력됩니다.

 

 

system call

 

I/O Multiplexing은 구현 방식과 성능에 따라 select system call과 epoll system call로 구분할 수 있습니다. select는 단순한 구현을 제공하지만, 성능이 상대적으로 낮고 fd 수가 많을 경우 비효율적입니다. 반면, epoll은 대규모 fd를 효율적으로 처리할 수 있는 고성능 메커니즘으로 설계되었으며, 특히 많은 클라이언트 연결을 처리할 때 뛰어난 성능을 발휘합니다.

 

》 select

• 대부분의 OS에서 지원하며, 범용적으로 사용 가능하다.
• fd_set에 fd를 등록하고 이 fd_set을 system call을 통해 확인한다.
• fd를 하나씩 확인하므로 시간이 선형적으로 증가하여 O(n)의 시간복잡도를 가진다.
• fd를 최대 1024개만 등록 가능하며, fd_set을 애플리케이션에서 직접 다루기 때문에 유지보수가 어렵다.

 

》 epoll

• 리눅스에서만 지원한다.
• epoll_create로 epoll 객체를 생성하고, epoll_ctl로 fd를 등록한 후, epoll_wait를 사용해 이벤트를 확인한다.
• 등록된 파일 디스크립터 중 이벤트가 발생한 fd만 반환되므로, 시간이 O(1)로 일정하게 유지된다.
• fd 수에 제한이 없으며, 대규모 fd를 효율적으로 처리할 수 있다.

 

 

Java NIO와 Selector

 

I/O Multiplexing에 대해 알아보았으니 이번에는 Java에서 I/O Multiplexing이 어떻게 구현되고 사용되는지 살펴보겠습니다. 이를 이해하기 위해서는 먼저 Java NIO에 대한 기본적인 이해가 필요합니다.

 

 

Java NIO

 

Java NIO는 Java 1.4에서 처음 도입된 기술로 Java New Input/Output의 약자입니다.

기존의 Java IO가 byte 또는 character 단위로 데이터를 처리하는 방식이었던 것과 달리, Java NIO는 버퍼(Buffer) 기반의 Channel과 Selector를 도입하여 Non-Blocking I/O를 지원하고, 높은 성능을 보장합니다.

 

이번 포스팅에서는 Java NIO의 여러 기술 중 SelectableChannel에 대해서만 살펴보겠습니다.

 

 

Java NIO에서 사용되는 SocketChannel 과 ServerSocketChannel은 모두 SelectableChannel을 상속하고 있습니다.

SelectableChannel은 Non-Blocking 모드로 전환할 수 있는 configureBlcoking()과 SelectableChannel을 Selector에 등록할 수 있는 register()라는 두 가지 중요한 메서드를 제공합니다.

 

Java NIO는 모두 Non-Blocking 하게 동작할 수 있다 (X)
FileChannel의 경우 SelectableChannel을 상속하고 있지 않기 때문에 Non-Blocking하게 동작할 수 없습니다.

 

 

 

Selector

 

위에서 살펴본 SelectableChannel의 메서드를 사용하면 I/O Multiplexing을 가능하게 해주는 Selector를 구현할 수 있습니다. I/O Multiplexing은 내부적으로 운영 체제의 system call을 통해 구현되며, 운영 체제에 따라 Selector의 구현체로 PollSelector, EPollSelector, KQueueSelector, WindowsSelector 등이 사용됩니다.

즉, Selector를 사용하면 여러 개의 Channel을 단일 쓰레드에서 효율적으로 관리할 수 있도록 설계할 수 있습니다.

 

 

java.nio.channels 패키지에 포함되어 있으며, 이벤트가 발생한 Channel이 있는지 확인하는 메서드인 select()와 이벤트가 발생한 Channel들의 정보를 담고 있는 SelectionKey 객체의 Set을 반환하는 selectedKeys() 라는 두 가지 중요한 메서드를 제공합니다.

 

》 Selector 생성 및 등록

@SneakyThrows
public static void main(String[] args) {
    log.info("start main");
    try (ServerSocketChannel serverChannel = ServerSocketChannel.open();
         Selector selector = Selector.open();) {
        serverChannel.bind(new InetSocketAddress("localhost", 8080));
        serverChannel.configureBlocking(false);
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            selector.select();
            Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();

            while (selectionKeys.hasNext()) {
                SelectionKey key = selectionKeys.next();
                selectionKeys.remove();

                if(key.isAcceptable()) {
                    SocketChannel clientChannel = ((ServerSocketChannel)(key.channel())).accept();
                    clientChannel.configureBlocking(false);
                    clientChannel.register(selector, SelectionKey.OP_READ);
                } else if(key.isReadable()) {
                    SocketChannel clientChannel = (SocketChannel) key.channel();

                    String requestBody = handleRequest(clientChannel);
                    sendResponse(clientChannel, requestBody);
                }
            }
        }
    }
}

@SneakyThrows
private static String handleRequest(SocketChannel clientChannel) {
    ByteBuffer requestByteBuffer = ByteBuffer.allocateDirect(1024);
    while (clientChannel.read(requestByteBuffer) == 0) {
    	log.info("Reading...");
    }
    requestByteBuffer.flip();
    String requestBody = StandardCharsets.UTF_8.decode(requestByteBuffer).toString();
    log.info("request : {}", requestBody);

    return requestBody;
}

@SneakyThrows
private static void sendResponse(SocketChannel clientChannel, String requestBody) {
    Thread.sleep(10);

    String content = "received : " + requestBody;
    ByteBuffer responseByteBuffer = ByteBuffer.wrap(content.getBytes());
    clientChannel.write(responseByteBuffer);
    clientChannel.close();
}

 

Selector.open() 메서드를 사용하여 Selector를 생성할 수 있으며, Closable을 구현했기 때문에 직접 close하지 않고, try-with-resources를 사용 가능합니다.

 

위에서 알아보았던 SelectableChannel의 configureBlocking() 메서드를 사용해 Non-Blocking 모드로 전환해 주고, register() 메서드를 사용하여 ServerSocketChannel에 Selector와 관심 있는 이벤트를 등록할 수 있습니다. 여기에서 관심있는 이벤트는, READ, WRITE, ACCEPT, CONNECT를 의미하며 SelectionKey라는 추상 클래스에서 int의 형태의 상수로 제공하고 있습니다.

 

또한 SelectionKey 클래스에서는 isReadable(), isWritable(), isConnectable(), isAcceptable() 이라는 이벤트 확인 메서드를 제공하며, 내부에 등록했던 Channel과 Selector도 포함하고 있습니다.

 

---

 

처음에는 ServerSocketChannel에 Selector와 함께 ACCEPT 이벤트를 등록합니다.

그 후 무한 루프를 돌면서 위에서 확인했던 Selector의 select() 메서드를 통해 Selector에서 이벤트가 발생한 채널이 있는지 확인합니다. 이때 반환 값은 이벤트가 발생한 Channel의 개수이며, 만약 Channel의 개수가 1개 이상이라면 selectionKeys() 메서드를 통해 이벤트가 발생한 Channel들의 정보를 담고 있는 SelectionKey 객체를 Set으로 받아오고 이를 순회하여 한 개씩 꺼내옵니다.

 

그 후, Set에서 꺼내온 SelectionKey에 등록된 채널이 어떠한 이벤트가 발생했는지 확인합니다.

1. ACCEPT 이벤트 발생
2. READ 이벤트 발생

만약 ACCEPT 이벤트가 발생했다면 SelectionKey로부터 ServerSocketChannel을 가져와 appcet()하여 SocketChannel을 받아오고, 받아온 SocketChannel에 동일한 Selector와 함께 READ 이벤트를 등록합니다.

 

만약 READ 이벤트가 발생했다면 SelectionKey로부터 SocketChannel을 가져와 데이터를 읽은 후 응답하고 SocketChannel을 close() 합니다.

 

 

Selector와 epoll

 

앞에서 알아본 Selector와 epoll은 비슷한 Event-driven 방식의 system call을 사용하여 I/O Multiplexing을 처리하기 때문에 동작을 다음과 같이 매칭해 볼 수 있습니다.

 

 

Reactor Pattern

 

 

Reactor Pattern은 동시에 들어오는 요청을 처리하는 이벤트 핸들링 패턴입니다. 이 패턴은 여러 이벤트 소스를 효율적으로 처리하기 위해 Selector와 같은 컴포넌트를 활용하여 요청을 비동기적으로 다룹니다.

 

Reactor Pattern을 그림처럼 표현할 수 있으며, Selector는 Demultiplexer 역할을 수행하여 들어오는 요청들을 적절한 Handler에게 동기적으로 전달합니다. 이때, Acceptor와 Handler는 모두 EventHandler 인터페이스의 구현체입니다. 다만, accept()를 담당하는 Acceptor는 다른 처리 로직을 따로 구분하여 관리하는 경우가 많기 때문에 구분하여 표현하였습니다.

 

위 내용을 보면 Selector에 대한 의문이 생기실 수도 있습니다.

그러면 Selector는 Multiplexer 인가? Demultiplexer 인가?

 

 

정답은 둘 다 맞다는 것입니다.

실제로 Selector 클래스의 주석에서는 Multiplexer라고 표현되어 있고, Reactor Pattern의 설명에서는 Demultiplexer라고 언급됩니다. 이는 Selector가 두 가지 중요한 역할을 동시에 수행하기 때문입니다.

• Multiplexer로서, Selector는 여러 채널을 동시에 처리할 수 있게 해주는 기술을 제공합니다.
• Demultiplexer로서, Selector는 각 채널에서 발생한 이벤트를 감지하고, 이를 적절한 처리 로직(핸들러)으로 분배하는 역할을 합니다.

따라서, Selector는 Multiplexer이자 Demultiplexer로서 두 가지 기능을 모두 수행한다고 할 수 있습니다.

 

 

Netty로 Reactor Pattern 구현하기

 

서론에서 I/O Multiplexing의 대표적인 예시로 언급했던 Netty도 Selector를 기반으로 구현되어 있기 때문에

앞서 배운 Reactor Pattern을 활용하여 Netty에서 사용되는 EventLoop를 간단하게 구현해 보겠습니다.

public static void main(String[] args) {
    List<EventLoop> eventLoopGroup = List.of(new EventLoop(8080));
    eventLoopGroup.forEach(EventLoop::run);
}

public class EventLoop {
    private final Logger log = LoggerFactory.getLogger(EventLoop.class);
    private final Selector selector;
    private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    public EventLoop(int port) throws IOException {
        selector = Selector.open();
        var serverSocket = ServerSocketChannel.open();
        serverSocket.bind(new InetSocketAddress("localhost", port));
        serverSocket.configureBlocking(false);
        serverSocket.register(selector, SelectionKey.OP_ACCEPT).attach(new Acceptor(selector));
    }

    public void run() {
        executorService.submit(() -> {
            log.info("Start EventLoop");
            while (true) {
                selector.select();
                Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();

                while (selectionKeys.hasNext()){
                    SelectionKey key = selectionKeys.next();
                    selectionKeys.remove();

                    dispatch(key);
                }
            }
        });
    }

    private void dispatch(SelectionKey key) throws IOException {
        EventHandler eventHandler = (EventHandler) key.attachment();

        if(key.isAcceptable() || key.isReadable()) {
            eventHandler.handle(key);
        }
    }
}

public interface EventHandler {
    void handle(SelectionKey key) throws IOException;
}

 

위 코드에 EventHandler의 구현체인 Acceptor와 Handler를 추가하면 완성됩니다.

완성된 코드는 GitHub에 업로드되어 있으니, 참고용으로 첨부하겠습니다.

 

 

마치며

 

이번 글에서는 I/O Multiplexing과 Selector에 대해 살펴보았습니다. Selector에 대한 이해를 돕기 위해 File Descriptor와 System Call에 대해서도 간단히 다뤄보았더니, 내용이 조금 길어졌습니다. 다음 글에서는 Reactor 패턴의 단점을 해결하기 위해 등장한 Proactor 패턴에 대해 알아보겠습니다.