[java] 자바 비동기 IO를 활용한 메시지 큐 개발하기

04 Dec 2023

java

메시지 큐는 분산 시스템에서 메시지를 안전하게 전달하는 중요한 컴포넌트입니다. 이번 블로그 포스트에서는 자바의 비동기 IO 기능을 활용하여 메시지 큐를 개발하는 방법을 알아보겠습니다.

비동기 IO란?

비동기 IO는 입출력 작업을 처리하는 동안 다른 작업을 동시에 수행할 수 있게 해주는 기술입니다. 기존의 동기식 IO 방식은 입출력 작업이 완료될 때까지 다음 코드로 진행되지 않았지만, 비동기 IO 방식은 작업을 시작한 후 결과를 기다리지 않고 다음 코드를 실행할 수 있습니다.

자바에서는 NIO(Non-blocking IO) 패키지를 통해 비동기 IO를 구현할 수 있습니다. NIO는 채널 기반의 IO 처리 방식인데, 이를 활용하여 메시지 큐를 개발할 것입니다.

메시지 큐 개발하기

다음은 메시지 큐를 개발하기 위한 간단한 예제 코드입니다.

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class MessageQueue {
    private AsynchronousServerSocketChannel serverSocketChannel;

    public MessageQueue(int port) throws Exception {
        serverSocketChannel = AsynchronousServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(port));
    }

    public void start() {
        serverSocketChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
            @Override
            public void completed(AsynchronousSocketChannel clientSocketChannel, Void attachment) {
                // 클라이언트 연결 완료됨
                process(clientSocketChannel);

                // 다음 연결을 받기 위해 accept를 다시 호출
                serverSocketChannel.accept(null, this);
            }

            @Override
            public void failed(Throwable exc, Void attachment) {
                // 연결 실패 처리
            }
        });
    }

    private void process(AsynchronousSocketChannel clientSocketChannel) {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        clientSocketChannel.read(buffer, null, new CompletionHandler<Integer, Void>() {
            @Override
            public void completed(Integer bytesRead, Void attachment) {
                if (bytesRead > 0) {
                    byte[] messageBytes = new byte[bytesRead];
                    buffer.flip();
                    buffer.get(messageBytes);

                    // 메시지 처리
                    String message = new String(messageBytes);
                    System.out.println("Received message: " + message);

                    // 다음 메시지를 읽기 위해 read를 다시 호출
                    buffer.clear();
                    clientSocketChannel.read(buffer, null, this);
                } else {
                    // 클라이언트 연결이 종료됨
                    try {
                        clientSocketChannel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }

            @Override
            public void failed(Throwable exc, Void attachment) {
                // 읽기 실패 처리
            }
        });
    }
}

위의 예제 코드는 비동기 소켓을 활용하여 메시지를 읽는 간단한 메시지 큐를 구현한 것입니다. 새로운 클라이언트 연결이 발생하면 accept 메서드가 호출되고, 연결이 수락되면 process 메서드로 처리합니다. 이후에는 메시지를 읽고 처리하는 과정이 반복됩니다.

결론

이번에는 자바의 비동기 IO를 활용하여 메시지 큐를 개발하는 방법에 대해 알아보았습니다. 비동기 IO를 활용하면 입출력 작업이 블로킹되지 않고 비동기적으로 처리되므로, 더욱 효율적인 시스템을 구축할 수 있습니다.