[Netty] 네티의 주요특징 - 동기와 비동기, 블로킹과 논블로킹

자바 네트워크 소녀 네티( 정경석 저 / 이희승 감수, 한빛미디어 ) 를 읽고 정리합니다. 

 

 

📘1부 2장. 네티의 주요특징 - 동기와 비동기, 블로킹과 논블로킹

 

🦋Netty

https://netty.io/

Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients.

비동기 이벤트 기반 네트워크 애플리케이션 프레임워크

유지보수를 고려한 고성능 프로토콜 서버와 클라이언트를 함께 개발할 수 있다.

 

🔹동기 비동기

동기 / 비동기/ 블로킹/ 논블로킹/ @Async, @EnableAsync과 관련된 내용은 다음 링크에 정리했다.

⇒ https://88dldl.tistory.com/162

 

비동기 호출을 지원하는 디자인패턴은 퓨처패턴, 옵저버 패턴, 리액터 패턴 등이 있다.

네티는 비동기 호출을 위한 api들을 프레임워크 레벨에서 제공하므로 개발자가 스레드 동기화 이슈 및 그에 따른 버그를 신경 쓰는 대신 구현할 기능에 집중할 수 있다.

 

 

🔹블로킹/ 논브로킹

블로킹 : 요청한 작업이 성공하거나 에러가 발생하기 전까지는 응답을 돌려주지 않는것

논블로킹 : 요청한 작업의 성공 여부와 상관없이 바로 결과를 돌려주는 것

 

소켓도 입출력 채널의 하나로서 NIO API를 사용할 수 있으며, 이것을 통해서 블로킹과 논블로킹 모드의 소켓을 사용할 수 있다.

 

🔸블로킹 소켓

블로킹 소켓과 논블로킹 소켓은 데이터 송수신을 위한 함수의 동작 방식에 따른 분류다.

자바는 두가지 소켓을 구분하기 위해 별도의 클래스를 사용한다.

 

블로킹 소켓 : ServerSocket, Socket 클래스

논블로킹 소켓 : ServerSocketChannel, SocketChannel 클래스

BlockingServer

public class BlockingServer {
    public static void main(String[] args) throws IOException {
        BlockingServer server = new BlockingServer();
        server.run();
    }

    public void run() throws IOException {
        ServerSocket server = new ServerSocket(8888); //1
        System.out.println("접속 대기중");

        while (true) {
            Socket sock = server.accept(); //2
            System.out.println("클라이언트 연결됨");
		
            OutputStream out = sock.getOutputStream(); //3
            InputStream input = sock.getInputStream(); //4

            while (true) {
                try {
                    int request = input.read(); //5
                    out.write(request);
                } catch (IOException e) {
                    break;
                }
            }
        }
    }
}

연결되는 클라이언트가 없으면 프로그램은 아무런 동작을 하지 않고, 스레드 또한 해당 함수의 완료를 기다리며 해당 위치에서 대기한다.

 

텔넷으로 접속하면 다음과 같이 뜬다.

서버에 접속하면 2에서 accept메서드의 실행이 완료되고 클라이언트와 연결된 소켓을 생성한다.

 

그리고 데이터를 읽는 read에서 다시 멈추고, 데이터의 입력을 기다린다.

⇒ 즉, 클라이언트로부터 데이터가 수신되기를 기다리며 스레드가 블로킹 된다.

 

입출력 메서드의 처리가 완료될 때까지 응답을 돌려주지 않고 대기한다.

 

클라이언트가 서버로 연결 요청을 보내면 서버는 연결을 수락하고 클라이언트와 연결된 소켓을 새로 생성하는데,

이때 메서드가 처리가 완료되기 전까지 스레드의 블로킹이 발생한다.

 

클라이언트가 전송한 데이터를 읽기 위해 read 메서드를 호출하고, 이 메서드의 처리가 완료되기 전까지 스레드가 블로킹 된다.

 

소켓으로 데이터를 전송하기 위해 write 메서드를 호출할 때도 블로킹이 발생하는데, write 메서드는 운영체제의 송신 버퍼에 전송할 데이터를 기록한다.

이때 송신 버퍼의 남은 크기가 write 메서드에서 기록한 데이터의 크기보다 작다면 송신 버퍼가 비워질 때까지 블로킹 한다.

 

이처럼 블로킹 소켓은 데이터 입출력에서 스레드의 블로킹이 발생하기 때문에 동시에 여러 클라이언트에 대한 처리가 불가능하게 된다.

 

이런 다중 클라이언트의 접속 처리를 하지 못하는 문제점을 해결하기 위해 연결된 클라이언트별로 각각 스레드를 할당하는 모델이 등장했다.

 

 

 

연결 당 스레드 생성 모델

해당 그림은 연결 당 스레드 생성 모델이다.

 

클라이언트가 서버에 접속하면 서버 소켓의 accept 메서드를 통해서 연결된 클라이언트 소켓을 얻어온다.

이때 블로킹 소켓은 I/O 처리에 블로킹이 발생하기 떄문에 새로운 스레드를 하나 생성하고 그 스레드에게 클라이언트 소켓에 대한 I/O 처리를 넘겨주게된다.

 

=> 서버 소켓이 동작하는 스레드는 다음 클라이언트의 연결을 처리할 수 있게 되는 것이다.

 

하지만 accept 지점이 병목 지점이다. accept 메서드는 단위 시간에 하나의 연결만을 처리하는 블로킹 모드로 동작하기 때문에 여러 클라이언트가 동시에 접속 요청을 하는 상황에서는 대기 시간이 길어진다.

또한 서버에 접속하는 클라이언트 수가 증가하면 애플리케이션 서버의 스레드 수가 증가해 자바 힙 메모리 부족으로 인한 Out Of Memory 오류가 발생할 수 있다.

 

이러한 상황이 발생하지 않도록 서버에서 생성되는 스레드 수를 제한 하는 방법인 스레드 풀링을 사용하기도 한다.

 

 

스레드 풀링 모델

 

 

해당 그림은 스레드 풀링 모델이다.

 

클라이언트가 서버에 접속하면 서버 소켓으로부터 클라이언트 소켓을 얻어온다.

다음으로 스레드 풀에서 가용 스레드를 하나 가져오고 해당 스레드에 클라이언트 소켓을 할당한다.

 

=> 이후에는 클라이언트 소켓에서 발생하는 I/O 처리를 할당된 스레드가 전담하게 된다.

 

이와 같은 구조는 동시에 접속 가능한 사용자 수가 스레드 풀에 지정된 스레드 수에 의존하게 된다.

 

 

|❓동시 접속 수를 늘리기 위해 스레드 풀의 크기를 자바 힙이 허용하는 최대 한도에 도달할 때까지 늘리는 것이 합당할까?

두가지의 관점에서 살펴보자

1. 자바의 가비지 컬렉션
   자바 프로세스는 가비지 컬렉션을 완료하기 위해 다른 스레드를 멈추게 된다.
   ⇒ 이때 애플리케이션이 먹통이 된 것처럼 보이게 되는데 힙 크기가 크면 클수록 가비지 컬렉션에 드는 시간이 길어진다.
   => 힙에 할당된 메모리가 크면 클수록 가비지 컬렉션이 수행되는 횟수는 줄어들지만 수행시간은 상대적으로 길어진다.
2. 운영체제에서 사용되는 컨텍스트 스위칭
   컨텍스트 스위칭이란 한 프로세스에서 수행되는 스레드들이 CPU 점유를 위해 자신의 상태를 변경하는 작업을 말한다. 
   이때 수많은 스레드가 CPU 자원을 획득하기 위해 CPU 자원을 소모하기 때문에 실제로 작업에 사용할 cpu자원이 적어지게 된다.

⇒ 블로킹 서버를 사용한 서버는 충분히 많은 동시 접속 사용자를 수용하지 못한다.

 

이러한 방식을 개선한 방식이 논블로킹 소켓 방식이다.

 

 

🔸논블로킹 소켓

논블로킹 모드의 소켓에서 데이터를 읽는 명령인 read메서드를 호출했다고 할때, 만약 클라이언트가 데이터를 아직 전송하지 않았거나 데이터가 수신버퍼까지 도달하지 않았다면 read 메서드를 읽어들인 바이트 길이인 0을 돌려준다.

NonBlockingServer

public class NonBlockingServer {
    private Map<SocketChannel, List<byte[]>> keepDataTrack = new HashMap<>();
    private ByteBuffer buffer = ByteBuffer.allocate(2 * 1024);

    private void startEchoServer() {
        try ( 
                Selector selector = Selector.open(); //2
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); //3
        ) {
            if ((serverSocketChannel.isOpen()) && (selector.isOpen())) {
                serverSocketChannel.configureBlocking(false);//5
                serverSocketChannel.bind(new InetSocketAddress(8888));//6

                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//7
                System.out.println("접속 대기중");

                while (true) {
                    selector.select();//8
                    Iterator<SelectionKey> keys = selector.selectedKeys().iterator();//9

                    while (keys.hasNext()) {
                        SelectionKey key = keys.next();
                        keys.remove();//10

                        if (!key.isValid()) {
                            continue;
                        }
                        if (key.isAcceptable()) { //11
                            this.acceptOP(key, selector);
                        } else if (key.isReadable()) { //12
                            this.readOP(key);
                        } else if (key.isWritable()) { //13
                            this.writeOP(key);
                        }
                    }
                }
            } else {
                System.out.println("서버 소켓을 생성하지 못했습니다.");
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    ...
}

논블로킹 소켓은 구조적으로 소켓으로부터 읽은 데이터를 바로 소켓에 쓸 수 없다.

이를 위해서 각 이벤트가 공유하는 데이터 객체를 생성하고 그 객체를 통해서 각 소켓 채널로 데이터를 전송한다.

 

[2] 자바 NIO 컴포넌트 중 하나인 Selector는 자신에게 등록된 채널에 변경 사항이 발생했는지 검사하고 변경 사항이 발생한 채널에 대한 접근을 가능하게 한다. open()메서드를 통해 새로운 Selector 객체를 생성했다.

[3] 블로킹 소켓의 ServerSocket에 대응되는 논블로킹 소켓의 서버 소켓 채널을 생성한다. 블로킹 소켓과 다르게 소켓 채널을 먼저 생성하고 사용할 포트를 바인딩한다.

[5] 소켓 채널의 블로킹 모드의 기본 값은 true이다. false로 설정해 논블로킹 모드로 설정한다.

[6] 포트를 지정하고 생성된 ServerSocketChannel 객체에 할당했다.

⇒ 이 작업이 완료되면 지정한 포트로 클라이언트 연결을 생성할 수 있다.

[7] ServerSocketChannel 객체를 Selector 객체에 등록한다. Selector가 감지할 이벤트는 연결 요청에 해당하는 SelectionKey.OP_ACCEPT 이다.

[8] 등록된 채널에서 변경사항이 발생했는지 검사한다. 아무런 I/O이벤트도 감지되지 않으면 이 부분에서 블로킹이 발생한다. 만약 이때 블로킹을 피하고 싶으면 selectNow 메서드를 사용하자

[9] Selector 에 등록된 채널에서 I/O 이벤트가 발생한 채널들의 목록을 조회한다.

[10] I/O 이벤트가 발생한 채널들의 목록에서 동일한 이벤트가 감지되는 것을 방지하기 위해 조회된 목록에서 제거한다.

[11] 조회된 I/O이벤트가 연결요청인지 확인한다.

[12] 조회된 I/O이벤트가 데이터 수신인지 확인한다.

[13] 조회된 I/O이벤트가 데이터 쓰기 가능인지 확인한다.

private void acceptOP(SelectionKey key, Selector selector) throws IOException {
    ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();//14
    SocketChannel socketChannel = serverChannel.accept();//15
    socketChannel.configureBlocking(false);//16

    System.out.println("클라이언트 연결됨 : " + socketChannel.getRemoteAddress());

    keepDataTrack.put(socketChannel, new ArrayList<byte[]>());
    socketChannel.register(selector, SelectionKey.OP_READ);//17
}

private void readOP(SelectionKey key) {
    try {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        buffer.clear();
        int numRead = -1;
        try {
            numRead = socketChannel.read(buffer);
        } catch (IOException e) {
            System.err.println("데이터 읽기 에러!");
        }

        if (numRead == -1) {
            this.keepDataTrack.remove(socketChannel);
            System.out.println("클라이언트 연결 종료 : " + socketChannel.getRemoteAddress());
            socketChannel.close();
            key.channel();
            return;
        }
        byte[] data = new byte[numRead];
        System.arraycopy(buffer.array(), 0, data, 0, numRead);
        System.out.println(new String(data, "UTF-8") + "from" + socketChannel.getRemoteAddress());
        doEchoJob(key, data);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

private void writeOP(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();
    List<byte[]> channelData = keepDataTrack.get(socketChannel);
    Iterator<byte[]> its = channelData.iterator();

    while (its.hasNext()) {
        byte[] it = its.next();
        its.remove();
        socketChannel.write(ByteBuffer.wrap(it)); // ByteBuffer.wrap()은 byte[]을 요구
    }
    key.interestOps(SelectionKey.OP_READ);
}

private void doEchoJob(SelectionKey key, byte[] data) {
    SocketChannel socketChannel = (SocketChannel) key.channel();
    List<byte[]> channelData = keepDataTrack.get(socketChannel);
    channelData.add(data);

    key.interestOps(SelectionKey.OP_WRITE);

}

public static void main(String[] args) {
    NonBlockingServer main = new NonBlockingServer();
    main.startEchoServer();
}

[14] 연결 요청 이벤트가 발생한 채널은 항상 ServerSocketChannel 이므로 이벤트가 발생한 채널을 ServerSocketChannel 로 캐스팅한다.

[15] 연결을 수락하고 연결된 소켓 채널을 가져온다.

[16] 논블로킹 모드로 설정한다.

[17] 클라이언트 소켓 채널을 Selector에 등록하여 I/O이벤트를 감시한다.

 

 

실행해보면 

다음과 같이 Non-blocking I/O(NIO) 기반으로 동작하며, 여러 클라이언트와 동시 통신 가능한 것을 확인 할 수 있다. 

 

🔸동작방식 비교

블로킹 방식과 논블로킹 방식의 가장 큰 차이점은 I/O 처리 방법에 있다.