네티의 주요 특징

동기와 비동기

“동기”라는 단어는 업계에 따라 다양한 의미를 갖는다.
이 책에서는 함수 또는 서비스의 호출 방식에 관한 내용을 다룬다.

동기

특정 서비스에 요청을 보내면, 내부에서 모든 작업이 처리 완료된 후에 결과를 받을 수 있다.
쉬운 디버깅, 직관적인 흐름 추적이 가능하다.

비동기

특정 서비스에 요청을 보내면, 내부에서 작업 완료되기 전에 일단 응답을 보낸다.
이후 사용자가 다시 결과 확인을 통해 요청을 보내면 어느 과정을 처리하고 있는지 응답을 보내준다.
Java의 Future 패턴, 이벤트 리스너의 옵저버패턴, Node.js의 콜백 함수, Netty의 Reactor패턴은 모두 비동기 방식이다.
수행 시간, 프로그램 구조 등 다양한 고민이 필요하다.
네티는 비동기 호출을 위한 API들을 프레임워크 레벨에서 제공하여 스레드 동기화 이슈 및 버그에 대한 부담을 덜어준다.

블로킹과 논블로킹

소켓의 동작 방식
블로킹
- 요청한 작업이 성공하거나 에러가 발생하기 전까지 응답을 돌려주지 않는 것
논블로킹
- 요청한 작업의 성공 여부와 상관없이 바로 결과를 돌려주는 것
- 응답값에 의해 에러나 성공 여부를 판단한다.
- JDK 1.4 부터 Non-Blocking IO를 제공하기 시작했다.
NIO API를 통해 블로킹과 논블로킹 모드의 소켓을 사용할 수 있다.

블로킹 소켓

ServerSocket을 통해 서버를 연다. 클라이언트가 접속하지 않으면 accept() 메서드 line에서 대기하게 된다.
이부분을 수행하는 스레드는 무한 대기(블로킹)하게 된다.
클라이언트가 무언가 데이터를 보내지 않으면, InputrStream의 read() 부분에서 또다시 멈추게 된다.
블로킹 소켓은 결국 호출된 입출력 메서드의 처리가 완료될 때 까지 응답을 돌려주지 않고 대기한다.

ServerSocket server = new ServerSocket(8888);
System.out.println("접속 대기중");

while (true) {
  Socket sock = server.accept();
  System.out.println("클라이언트 연결됨");

  OutputStream out = sock.getOutputStream();
  InputStream in = sock.getInputStream();

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

위 코드 외에도 클라이언트가 소켓 채널에 write할 때 운영체제의 송신 버퍼의 크기가 전송할 데이터를 담을만큼 충분하지 않다면 송신 버퍼가 비워질 때 까지 블로킹된다.
블로킹 소켓은 데이터 입출력에서 스레드의 블로킹이 발생하여 동시에 여러 클라이언트를 처리하기에 적합하지 않다.
연결 당 스레드를 할당하는 방법
- 클라이언트 연결과 직결되는 accept() 메서드에서 병목이 발생할 수도 있고, 스레드가 증폭되다가 힙 메모리가 부족해져 OOM 에러가 발생할 수도 있다.
- OOM 에러를 피하기 위해 스레드 풀을 사용할 수도 있지만 동시 접속 클라이언트 수가 스레드 풀의 스레드 개수에 의존하게 된다.
- 동시 접속 수를 늘리기 위해 스레드 풀 크기를 자바 힙이 허용하는 최대 크기까지 늘리도록 하더라도, GC가 힙 메모리가 크면 수행시간이 길어져 애플리케이션의 중단 시간이 길어질 수 있고, 수많은 스레드가 CPU 자원을 획득하기 위해 경쟁하며 CPU 자원을 소모하게 된다.

논블로킹 소켓

소켓에서 데이터를 읽는 read 메서드를 호출했을 때, 블로킹 소켓은 클라이언트가 데이터를 전송해 수신버퍼에 데이터가 들어올 때 까지 블로킹된다.
논블로킹 소켓은 클라이언트가 데이터를 전송하지 않았거나 수신 버퍼에 데이터가 들어오지 않았다면 읽어들인 바이트 길이인 0을 돌려준다.
다소 복잡한 감이 있지만, 크게 나눠보자면 소켓 서버를 시작하는 startEchoServer 메서드, 연결을 맺는 acceptOP 메서드, I/O로부터 읽는 readOP 메서드, I/O에 쓰는 writeOP 메서드가 있다.

startEchoServer

Selector 클래스를 통해 자신에게 등록된 채널에 변경 사항이 발생했는지 검사하고 변경 사항이 발생한 채널에 대한 접근을 가능하게 한다.
ServerSocketChannel 클래스를 통해 소켓 서버를 연다. 소켓 채널을 먼저 생성한 후 포트를 바인딩한다.
Selector객체가 연결 요청인 SelectKey.OP_ACCEPT 이벤트를 감지하도록 소켓 채널에 등록한다.
Selector에 등록된 채널에서 변경사항이 발생하는지 검사한다. 만약 I/O 이벤트가 발생했다면 어떤 이벤트인지 확인하여 적절한 처리를 위해 메서드를 호출한다.
- accetable, readable, writable을 확인하고 acceptOp, readOp, writeOp 메서드를 각각 호출하는 부분이 이에 해당한다.

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();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()
    ) {

      if ((serverSocketChannel.isOpen()) && (selector.isOpen())) {
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.bind(new InetSocketAddress(8888));

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

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

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

            if (!key.isValid()) {
              continue;
            }

            if (key.isAcceptable()) {
              this.acceptOP(key, selector);
            } else if (key.isReadable()) {
              this.readOP(key);
            } else if (key.isWritable()) {
              this.writeOP(key);
            }
          }
        }
      } else {
        System.out.println("서버 소캣을 생성하지 못했습니다.");
      }
    }
    catch (IOException ex) {
      System.err.println(ex);
    }
  }

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

클라이언트의 연결을 수락하고 연결된 소켓 채널을 가져온다.
소켓 채널을 논블로킹으로 설정하고 Selector 에 등록하여 OP_READ 이벤트를 감시한다.

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

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

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

readOp, writeOp 생략

데이터 읽기, 쓰기 처리하는 메서드는 아래와 같이 구현된다.

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.cancel();
      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 ex) {
    System.err.println(ex);
  }
}

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);
}

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));
  }

  key.interestOps(SelectionKey.OP_READ);
}

이벤트 기반 프로그래밍

이벤트 추상화가 고수준이면 세부적인 제어가 힘들고, 저수준이면 한 동작에 대해 너무 많은 이벤트가 발생해 애플리케이션 성능이 저하된다.
서버에 연결된 클라이언트들의 경우 이벤트의 추상화에 대해 깊은 고민이 필요하다. 연결할 클라이언트 수나 이벤트 수가 매우 가변적이며 예측이 불가능하기 때문이다.
네트워크 프로그램에서 이벤트가 발생하는 주체는 소켓 이다.
- 소켓이란 데이터 송수신을 위한 네트워크 추상화 단위로, 일반적으로 네트워크 프로그램에서 소켓은 IP와 포트를 가지고 있으며 양방향 네트워크 통신이 가능한 객체이다.
- 소켓에 데이터를 읽고쓰려면 NIO(소켓 채널) 혹은 스트림(OIO, Old Blocking IO)를 사용해야 한다.
- 클라이언트 애플리케이션이 소켓에 연결된 스트림에 데이터를 쓰면, 서버에 데이터가 전송된다.
발생하는 이벤트는 소켓 연결, 데이터 송수신 으로 나눌 수 있다.
Netty는 데이터의 읽고 쓰기를 위한 이벤트 핸들러(혹은 데이터 핸들러)인 ChannelInboundHandlerAdapter 를 제공한다.
사용자는 데이터를 소켓으로부터 직접 읽고 쓰지 않고 이벤트 핸들러를 통해 읽고 쓸 수 있다.
서버 애플리케이션의 코드를 클라이언트 애플리케이션에서 재사용할 수 있으며, 각 이벤트에 따라 로직을 분리할 수도 있다.
에러 이벤트도 같이 정의하므로, 예외 처리가 쉽다.

Previous네티 맛보기 Next채널 파이프라인

Last updated 7 months ago