[Netty] Start Netty Programming 1 - 개요

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

    private void run() throws IOException {
        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 {
                    // blocking method, 운영체제의 송신버퍼에 전송할 데이터를 기록한다. 
                    // 이때 송신버퍼의 남은 크기가 write 메소드에서 기록한 데이터의 크기보다 작다면 
                    // 송신버퍼가 비워질 때까지 블로킹된다.
                    int request = in.read();
                    out.write(request);
                }
                catch (IOException e) {
                    break;
                }
            }
        }
    }
}

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 = Selector.open();

          // blocking 소켓의 ServerSocket에 대응하는 Non-Blocking 소켓클래스
          ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()
        ) {

          if ((serverSocketChannel.isOpen()) && (selector.isOpen())) {
             serverSocketChannel.configureBlocking(false); // default value is true. 비동기로 하려면 세팅필요
             serverSocketChannel.bind(new InetSocketAddress(8888)); // port binding

             // ServerSocketChannel 에 Selector 를 등록한다.
             // Selector가 감지할 이벤트는 연결 요청에 해당하는 Accept() Operation 이다.
             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
             System.out.println("접속 대기중");

             while (true) {
                // Selector 에 등록된 채널에서 변경사항이 있는지 검사한다.
                // Selector 에 아무런 I/O 이밴트도 발생하지 않으면 스레드는 이 부분에서 블로킹 된다.
                // I/O 이벤트가 발생하지 않았을 때 블로킹을 피하고 싶다면 selectNow()를 사용하면 된다.
                selector.select();

                // Selector 에 등록돤 채널 중에서 I/O 이벤트가 발생한 채널들의 목록을 조회한다.
                Iterator<SelectionKey> keys = selector.selectedKeys().iterator();

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

                   // I/O 이벤트가 발생한 채널에서 동일한 이벤트가 감지되는 것을 방지하기 위해 제거한다.
                   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);
       }
    }

    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[]>());

       // 클라이언트 소켓 채널을 Selector에 등록하여 I/O 이벤트를 감시한다.
       socketChannel.register(selector, SelectionKey.OP_READ);
    }

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

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