java.nio

서론

java.nio는 new input output의 약자로, 자바의 기존 i/o를 개선하기 위한 new i/o 입니다.

java.io & java.nio

java.io는 다음과 같은 문제를 가지고 있습니다.

• Blocking I/O
  • 데이터 read/write 시 스레드가 BLOCKED 상태가 되어 다른 작업을 수행할 수 없습니다.
• 스트림 기반
  • 스트림 기반이기 때문에 데이터 read/write 시 데이터를 순차적으로 처리합니다.
• 양방향 통신 불가능
  • 양방향 통신이 불가능하기 때문에, 각각 read와 write에 해당하는 스트림을 정의해야 합니다.

이를 java.nio는 다음과 같이 해결했습니다.

• Blocking I/O
  • 셀렉터(Selector)와 채널(Channel)을 활용해 스레드가 BLOCKED 상태가 되지 않고 여러 I/O 작업을 동시에 처리할 수 있는 Non-Blocking 기능을 지원합니다.
• 스트림 기반
  • 버퍼(Buffer)를 통해 데이터를 관리해, 스트림보다 빠르게 데이터를 처리할 수 있습니다.
  • 스트림 기반에서 버퍼 기반으로 변경하면서, 순차적으로 처리하던 데이터를 병렬적으로 처리할 수 있게 됩니다.
• 양방향 통신 가능
  • 하나의 버퍼를 통해 read/write 모두 가능합니다.

java.nio

구성 요소

java.nio에는 다음과 같이 핵심이 되는 개념이 존재합니다.

• Selector
  • 하나의 스레드가 여러 Channel의 I/O 상태를 모니터링할 수 있게 해 주는 구성 요소입니다.
  • Selector를 기반으로 이벤트 기반의 프로그래밍이 가능합니다.
  • SelectionKey
    • Selector와 Channel 사이의 관계를 나타냅니다.
    • 하나의 Channel이 Selector에 등록되면, 등록되는 즉시 SelectionKey 객체가 생성됩니다.
    • SelectionKey를 통해 현재 Channel의 상태를 확인하고, 이벤트 발생 시 그에 대한 처리가 가능합니다.
• Channel
  • I/O 작업을 위한 객체입니다.
  • 다양한 환경에 최적화된 여러 Channel을 제공합니다.
• Buffer
  • 데이터를 임시 저장하는 메모리 영역입니다.
  • 하나의 버퍼에서 read/write 모두 가능합니다.
    • 변경하기 위해 모드 및 버퍼 초기화가 필요합니다.

예제 & 동작 방식

java.nio의 동작 방식은 사실 멀티플렉싱 기반의 다중 서버 동작 방식과 유사합니다.
java.nio의 핵심 기능 중 하나가 멀티플렉싱 기반의 I/O이기 때문입니다.

위에 있는 그림이 멀티플렉싱 기반의 다중 서버의 동작 방식입니다.

이를 java.nio가 Non-Blocking I/O 방식으로 하나의 스레드가 여러 소켓을 수행하는 과정으로 변경해 동작 과정을 살펴보고자 합니다.

이 때, 동작 순서는 간단한 echo 서버 예제를 통해 확인해보고자 합니다.

public class EchoServer {

    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();

        ServerSocketChannel server = ServerSocketChannel.open();
        server.bind(new InetSocketAddress("localhost", 8080));
        server.configureBlocking(false);
        server.register(selector, SelectionKey.OP_ACCEPT);

        ByteBuffer buffer = ByteBuffer.allocate(256);

        while (true) {
            selector.select();

            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> iter = selectedKeys.iterator();

            while (iter.hasNext()) {
                SelectionKey key = iter.next();

                iter.remove();

                if (key.isAcceptable()) {
                    register(selector, server);
                }

                if (key.isReadable()) {
                    answerWithEcho(buffer, key);
                }
            }
        }
    }

    private static void answerWithEcho(ByteBuffer buffer, SelectionKey key) throws IOException {
        SocketChannel client = (SocketChannel) key.channel();

        int readBytes = client.read(buffer);

        if (readBytes == -1 || "EXIT".equals(new String(buffer.array()).trim())) {
            client.close();
            System.out.println("Server : 클라이언트와의 연결을 종료합니다.");
        }
        else {
            buffer.flip();
            client.write(buffer);
            buffer.clear();
        }
    }

    private static void register(Selector selector, ServerSocketChannel serverSocket) throws IOException {
        SocketChannel client = serverSocket.accept();

        client.configureBlocking(false);
        client.register(selector, SelectionKey.OP_READ);
        System.out.println("Server : 클라이언트와 연결되었습니다 : " + client.getRemoteAddress());
    }
}

public class EchoClient {

    public static void main(String[] args) throws IOException {
        SocketChannel client = SocketChannel.open(new InetSocketAddress("localhost", 8080));

        System.out.println("Client : 서버와 연결되었습니다 : " + client.getRemoteAddress());

        Scanner sc = new Scanner(System.in);

        while (true) {
            System.out.print("Client : 서버에 전달할 문자열을 입력해주세요 : ");
            String input = sc.nextLine();

            ByteBuffer buffer = ByteBuffer.wrap(input.getBytes());

            client.write(buffer);
            buffer.clear();
            client.read(buffer);

            String response = new String(buffer.array()).trim();

            if ("EXIT".equals(response)) {
                System.out.println("Client : 서버와의 연결을 종료합니다.");
                client.close();
                break ;
            }

            System.out.println("Client : 서버로부터의 응답 : " + response);
            buffer.clear();
        }
    }
}

1. 셀렉터에 I/O를 수행할 다양한 채널을 등록합니다.
   1-1. 이 때, Non-Blocking I/O 방식으로 동작해야 하므로 채널의 옵션을 Non-Blocking I/O로 변경합니다. 이후 ServerSocketChannel은 클라이언트와 연결된 소켓에 추가적인 설정을 진행해야 하므로 ServerSocketChannel을 등록할 때 이벤트 타입을 SelectionKey.OP_ACCEPT로 설정합니다.
   1-2. 소켓 통신을 수행하므로 ServerSocketChannel을 등록합니다. 해당 채널은 TCP 연결 요청이 들어올 때 까지 Listen하다가 연결 요청이 오면 이를 수락합니다. 수락하면, 클라이언트와 연결된 소켓을 가진 SocketChannel을 반환합니다.
2. 셀렉터에 채널을 등록하는 즉시, 채널의 상태를 모니터링 할 수 있는 셀렉션 키를 반환합니다.
3. 애플리케이션은 무한 반복문을 통해 셀렉터에게 데이터가 준비된 채널이 있는지 확인합니다.
4. 클라이언트가 서버 소켓으로 연결 요청을 보냅니다.
5. 셀렉터에 등록된 ServetSocketChannel의 ServetSocket이 클라이언트의 연결 요청을 감지하고 수락합니다.
   5-1. ServerSocket이 클라이언트의 연결 요청을 수락했다는 의미는 ServerSocket을 통해 I/O를 하기 위한 데이터가 준비되었음을 감지(클라이언트의 소켓 연결 요청 데이터 세팅)했음을 의미합니다. 이는 ServerSocketChannel가 I/O를 수행하기 위해 준비되었음을 의미합니다.
   5-2. 클라이언트와 연결된 SocketChannel을 Non-Blocking I/O로 설정합니다. 이후 클라이언트에서 전송한 데이터를 조회해야하므로 SocketChannel을 셀렉터에 등록할 때 이벤트 타입을 SelectionKey.OP_READ로 설정합니다.
6. 무한 반복문을 통해 준비된 채널을 감지하고 있던 애플리케이션이 셀렉터를 통해 준비된 채널들의 정보를 셀렉션 키를 통해 반환받고 발생한 이벤트에 따라 분기 처리해 필요한 로직을 수행합니다.
7. 클라이언트가 서버에게 데이터를 전송합니다.
   7-1. 전송한 데이터가 커널 영역에서 유저 영역으로 복사됩니다. 이는 전송한 데이터를 읽을 준비가 된 것이기 때문에, ServerSocketChannel가 I/O를 수행하기 위해 준비되었음을 의미합니다.
8. 무한 반복문을 통해 준비된 채널을 감지하고 있던 애플리케이션이 셀렉터를 통해 준비된 채널들의 정보를 셀렉션 키를 통해 반환받고 발생한 이벤트에 따라 분기 처리해 필요한 로직을 수행합니다.
9. 클라이언트가 전달한 데이터를 버퍼로 유저 영역에서 읽어옵니다. 데이터를 토대로 클라이언트의 요청을 처리하고, 그 결과를 버퍼에 write해 클라이언트에게 데이터를 전달합니다.
   9-1. 하나의 버퍼를 양방향으로 쓰기 위해, flip()을 활용해 모드를 변경해주어야 합니다.
10. 클라이언트가 서버에서 전송한 데이터를 버퍼로 읽습니다.

다음과 같이 예제가 동작하는 것을 확인할 수 있습니다.