Netty 전송 API

Ch04. 전송 API

네트워크를 통해 전송되는 데이터는 모두 바이트 형식이다.

 

네티는 모든 전송 구현에 공통 API를 기반 레이어로 활용하므로 JDK를 직접 이용할 때보다 블로킹-논블로킹 변환 작업이 훨씬 간단하게 이루어진다.

코드가 구현 세부사항으로 오염될 우려가 적고, 전체 코드 기반을 광범위하게 리팩터링할 필요가 없다.

 

 

연결을 수락한 다음, 클라이언트로 "Hi!" 라는 메시지를 전송한 뒤 연결을 닫는 간단한 애플리케이션을 작성해보자.

NettyNioServer

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.CharsetUtil;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

public class NettyNioServer {
    public static void main(String[] args) throws Exception{
        int port = 8888;
        final ByteBuf buf = Unpooled.copiedBuffer("Hi! \n", Charset.forName("UTF-8"));
        EventLoopGroup group = new NioEventLoopGroup();    // 논블로킹 모드를 위해 NioEventGroup을 이용
        try {
            ServerBootstrap b = new ServerBootstrap();    // ServerBootstrap 생성
            b.group(group).channel(NioServerSocketChannel.class)
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {    // 연결이 수락될 때마다 호출될 ChannelInitializer를 지정
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){    // 이벤트를 수신하고 처리할 ChannelInboundHandlerAdapter를 추가
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    ctx.writeAndFlush(buf.duplicate())
                                        .addListener(ChannelFutureListener.CLOSE);
                                }
                            });
                        }
                    });
            ChannelFuture f = b.bind().sync();
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

 

일반적으로 Netty를 사용하지 않은 구현에 비해 가독성이 높고, 간편하다는 것을 알 수 있다.

모든 구현들이 Channel, ChannelPipeline, ChannelHandler를 기준으로 정의되고 있음을 기억해두자.

 

---

 

 

Channel 인터페이스는 모든 입출력 작업에 이용되므로 전송 API의 핵심이라고 볼 수 있다.

Channel에는 ChannelPipeline과 ChannelConfig 등이 할당되는데,

ChannelConfig는 Channel에 대한 모든 구성 설정을 포함하며, 임시 변경(hot change)을 지원한다.

 

또, Channel은 고유하므로 정렬 순서를 보장하기 위해 java.lang.Comparable을 이용한다. 즉, 고유한 두 Channel 인스턴스가 동일한 해시코드를 반환하는 경우에는 Error가 발생한다.

 

 

ChannelPipeline은 인바운드와 아웃바운드 데이터와 이벤트에 적용될 ChannelHandler 인스턴스를 모두 포함한다.

일반적인 ChannelPipeline의 용도는 다음과 같다.

• 데이터를 한 포맷에서 다른 포맷으로 변환
• 예외에 대한 알림 제공
• Channel의 활성화 또는 비활성화에 대한 알림 제공
• Channel을 EventLoop에 등록할 때 또는 등록 해제할 때 알림 제공
• 사용자 정의 이벤트에 대한 알림 제공

 

 

ChannelHandler 인스턴스를 필요에 따라 추가하거나 제거해 ChannelPipeline을 즉석으로 수정할 수 있으며, 이러한 네티의 특징을 활용하면 아주 유연한 애플리케이션을 개발할 수 있다는 것을 기억하자.

 

 

 

원격 피어로 데이터를 기록하고 플러시하는 일반적인 Channel 작업의 예를 보자.

Channel channel = ...
ByteBuf buf = Unpooled.copiedBuffer("your data", CharsetUtil.UTF_8);
ChannelFuture cf = channel.writeAndFlush(buf);    // 데이터를 기록하고 플러시
cf.addListerner(new ChannelFutureListener(){    // 기록이 완료되면 알림을 받을 ChannelFutureListener를 추가
    @Override
    public void operationComplete(ChannelFuture future) {
        if(future.isSuccess()){    // 기록이 완료될 경우
            System.out.println("Write success!");
        } else {
            System.out.println("Write error");    // 오류를 로깅
            future.cause().printStacktrace();
        }
    }
})

 

 

네티의 Channel 구현은 스레드에 대해 안전하므로 여러 스레드를 이용하는 경우에도 Channel의 참조를 저장하고 원격 피어에 뭔가를 출력할 때 이용할 수 있다. 다음 예제에는 여러 스레드로 기록을 수행하는 간단한 예를 보자.

 

 

Channel channel = ...
ByteBuf buf = Unpooled.copiedBuffer("your data", CharsetUtil.UTF_8).retain();
Runnable writer = new Runnable() {    // 데이터를 채널로 기록하는 Runnable 생성
    @Override
    public void run() {
        channel.write(buf.duplicate());
    }
};

Executor executor = Executors.newCachedThreadPool();    //  스레드풀 Executor에 대한 참조를 얻음

// 한 스레드에 기록
executor.execute(writer);

// 다른 스레드에 기록
executor.execute(writer);

...

 

 

 

---

 

 

 

NIO

NIO는 모든 입출력 작업의 완전한 비동기 구현을 제공하며 셀렉터 기반 API를 활용한다.

셀렉터의 기본 개념은 Channel의 상태가 변경되면 요청이 알림을 받을 수 있는 레지스트리 역할을 하는 것이다.

지원되는 상태 변경은 다음과 같다.

• 새로운 Channel이 수락되고 준비됨.
• Channel 연결이 완료됨
• Channel에 읽을 데이터가 있음
• Channel을 이용해 데이터를 기록할 수 있음

 

 

애플리케이션이 상태 변경에 반응한 후에는 셀렉터가 재설정되며 해당 스레드에서 변경을 검사하고 적절하게 반응하는 프로세스가 반복된다.

 

 

 

 

 

논블로킹 코드 기반의 환경에서..

코드 기반에 블로킹 호출이 없거나 블로킹 호출을 제한할 수 있다면 NIO 또는 epoll(리눅스의 경우)을 이용하는 것이 좋다. 다수의 동시 연결을 처리하기 위한 것이지만 동시 연결이 많지 않은 경우에도 아주 잘 작동한다. 특히 여러 연결에 스레드를 공유할 수 있다는 것이 장점이다.

 

 

 

 

 

출처 : 네티인 액션

https://book.naver.com/bookdb/book_detail.nhn?bid=10462610

네티 인 액션