[JAVA - NIO] Netty

Netty

: Netty는 Java로 작성된 비동기 이벤트 기반의 네트워크 애플리케이션 프레임워크입니다. 네트워크와 관련된 다양한 작업을 쉽게 처리할 수 있도록 설계되었습니다. Netty는 고성능, 확장성, 유연성, 다양한 프로토콜 지원 등의 특징을 갖고 있습니다.

 

1. 비동기 이벤트 기반: Netty는 이벤트 기반의 비동기 프로그래밍 모델을 사용하여 네트워크 작업을 처리합니다. 이를 통해 블로킹하지 않고 다중 클라이언트를 효율적으로 처리할 수 있습니다.
   
   
2. 고성능: Netty는 고성능을 제공하기 위해 설계되었습니다. 네트워크 작업에 대한 최적화된 IO 처리와 스레드 관리를 통해 높은 처리량과 낮은 지연 시간을 보장합니다.
   
   
3. 유연성: Netty는 다양한 프로토콜을 지원하며, 커스터마이징이 쉽습니다. HTTP, HTTPS, TCP, UDP 등의 프로토콜을 지원하며, 이를 활용하여 다양한 네트워크 애플리케이션을 개발할 수 있습니다.

 

1.  셀렉터(Selector)

: Selector는 다중 채널을 모니터링하고, 그 중 입출력이 가능한 채널을 선택할 수 있는 기능을 제공합니다. 이를 통해 단일 스레드에서 여러 채널을 관리하고 다중 연결을 효율적으로 처리할 수 있습니다.

 

 

2. 이벤트(Event)

1. ChannelActive : 채널이 활성화될 때 발생하는 이벤트입니다. 채널이 연결되었거나 채널에 입출력 작업이 준비된 경우에 발생합니다.
   
   
2. ChannelInactive : 채널이 비활성화될 때 발생하는 이벤트입니다. 채널이 닫히거나 연결이 끊어진 경우에 발생합니다.
   
   
3. ChannelRead : 채널에서 데이터를 읽을 때 발생하는 이벤트입니다. 채널에 데이터가 도착했을 때 발생하며, 읽은 데이터를 처리하는 작업을 수행할 수 있습니다.
   
   
4. ChannelReadComplete : 채널에서 데이터 읽기가 완료되었을 때 발생하는 이벤트입니다. 채널에서 모든 데이터를 읽었을 때 발생하며, 이벤트 핸들러에서 데이터 처리가 완료되었음을 알리는 데 사용됩니다.
   
   
5. ChannelWritabilityChanged : 채널의 쓰기 가능 상태가 변경될 때 발생하는 이벤트입니다. 채널의 출력 버퍼가 가득 차거나 비어있을 때 발생합니다.
   
   
6. UserEvent: 사용자 정의 이벤트입니다. 개발자가 직접 이벤트를 정의하고 발생시킬 수 있습니다.

 

3. Netty핵심 컴포넌트

 

컴포넌트	설명
Channel	인바운드 아웃바운드 운송수단
콜백 (Callback)	인바운드, 아웃바운드 핸들러에서 사용 channelRegistered : 채널이 이벤트 루프에 등록될 때 호출됩니다. channelUnregistered : 채널이 이벤트 루프에서 등록 해제될 때 호출됩니다. channelActive : 채널이 활성화될 때 호출됩니다. 이는 채널이 연결되었거나 준비된 상태임을 의미합니다. channelInactive : 채널이 비활성화될 때 호출됩니다. 이는 채널이 닫히거나 연결이 끊어진 상태임을 의미합니다. channelRead : 채널에서 데이터를 읽을 때 호출됩니다. channelReadComplete : 채널에서 데이터 읽기가 완료되었을 때 호출됩니다. channelWritabilityChanged : 채널의 쓰기 가능 상태가 변경될 때 호출됩니다. exceptionCaught : 예외가 발생했을 때 호출됩니다.
Future	비동기 작업의 결과를 저장
이벤트와 핸들러	이벤트그룹  : 이벤트 그룹은 여러 개의 이벤트 루프를 포함하고 있으며, 각각의 이벤트루프는 하나의 쓰레드에서 동작합니다. 이벤트 그룹은 주로 두 가지 종류로 나뉩니다: 부모 이벤트 루프 (Parent EventLoop): 클라이언트와 서버를 연결하고 수락하는 작업을 수행하는 이벤트 루프입니다. 이벤트 루프 그룹에 하나의 부모 이벤트 루프만 포함되어 있는 경우가 일반적입니다. 자식 이벤트 루프 (Child EventLoop): 연결된 채널에서 발생하는 모든 I/O 작업을 처리하는 이벤트 루프입니다. 이벤트 루프 그룹에는 여러 개의 자식 이벤트 루프가 포함될 수 있으며, 이는 다중 연결을 효과적으로 처리하기 위한 것입니다. 파이프라인 :  네트워크 작업을 처리하는 여러 개의 핸들러를 연결하고, 이를 통해 네트워크 이벤트를 처리하고 데이터를 변환하는 역할을 수행합니다. 핸들러 : 인바운드 핸들러 (Inbound Handler): 인바운드 핸들러는 채널로 들어오는 데이터를 처리하는 역할을 담당합니다. 주로 데이터를 읽고 해석하거나, 연결 수락과 같은 네트워크 이벤트를 처리합니다. ChannelInboundHandler 인터페이스를 구현하여 작성됩니다. 아웃바운드 핸들러 (Outbound Handler): 아웃바운드 핸들러는 채널에서 나가는 데이터를 처리하는 역할을 합니다. 주로 데이터를 변환하거나, 특정 프로토콜에 따라 데이터를 가공하는 등의 작업을 수행합니다. ChannelOutboundHandler 인터페이스를 구현하여 작성됩니다.

 

 

하나의 이벤트 루프에서 두 개의 채널에서 WRITE 이벤트가 발생하면, 네티의 이벤트 루프는 순차적으로 각 채널의 WRITE 이벤트를 처리합니다.

기본적으로 네티의 이벤트 루프는 단일 쓰레드에서 실행되기 때문에, 이벤트 루프가 한 번에 하나의 이벤트를 처리합니다. 따라서 두 개의 채널에서 WRITE 이벤트가 발생하면, 이벤트 루프는 한 번에 하나의 채널의 WRITE 이벤트를 처리하고 다음 채널의 WRITE 이벤트를 처리합니다.

 

Netty서버, 클라이언트

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.68.Final</version> <!-- 또는 최신 버전 -->
</dependency>

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.ChannelHandlerContext;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class NettyServer implements CommandLineRunner {

    @Override
    public void run(String... args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new SimpleChannelInboundHandler<String>() {
                 @Override
                 protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                     System.out.println("Received message: " + msg);
                     ctx.writeAndFlush("Hello from Netty Server\n");
                 }
             });

            ChannelFuture f = b.bind(8080).sync();
            System.out.println("Netty server started on port 8080");
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

 

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.ChannelHandlerContext;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class NettyClient implements CommandLineRunner {

    @Override
    public void run(String... args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.SO_KEEPALIVE, true)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
                         @Override
                         protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                             System.out.println("Received from server: " + msg);
                         }
                     });
                 }
             });

            ChannelFuture f = b.connect("localhost", 8080).sync();
            f.channel().writeAndFlush("Hello from Netty Client\n");
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

 

 

---

 

ByteBuf Class

• readByte(), writeByte(): 바이트를 읽고 씁니다.
• readBytes(int length), writeBytes(byte[] bytes): 바이트 배열을 읽고 씁니다.
• copy(): 버퍼의 복사본을 만듭니다. (새로운 객체)
• slice(): 부분 버퍼를 만듭니다.
• duplicate(): 현재 버퍼의 복사본을 만듭니다. (중복된 객체)
• release(): 버퍼의 메모리를 해제합니다.
• readerIndex() 메서드를 사용하여 현재 읽기 인덱스를 확인할 수 있습니다.
• writerIndex() 메서드를 사용하여 현재 쓰기 인덱스를 확인할 수 있습니다.
• setIndex(int readerIndex, int writerIndex) 메서드를 사용하여 읽기 인덱스를 설정할 수 있습니다.

 

1. 읽기 인덱스 (Reader Index):
   • 읽기 인덱스는 현재 읽을 위치를 나타냅니다.
   • read 메서드를 호출하면 읽기 인덱스가 이동하며, 읽은 데이터의 크기만큼 증가합니다.
   • 초기에는 0으로 설정되며, 데이터를 읽을 때마다 증가합니다.
2. 쓰기 인덱스 (Writer Index):
   • 쓰기 인덱스는 현재 쓸 위치를 나타냅니다.
   • write 메서드를 호출하면 쓰기 인덱스가 이동하며, 쓴 데이터의 크기만큼 증가합니다.
   • 초기에는 0으로 설정되며, 데이터를 쓸 때마다 증가합니다.

ByteBuf는 몇가지 특징을 더 가지고 있는데 inbound handler에서 사용되는 ByteBuf는 자동으로 release 하지 않으며 release를 직접 수행해야합니다. outbound handler에서 사용되는 ByteBuf는 write 한 이후에 자동적으로 release 수행합니다.

 

 

메모리 누수겁출 옵션

 

java -Dio.netty.leakDetection.level=ADVANCED

 

 

DISABLED	메모리 누수 검출 기능이 비활성화됩니다. 디폴트 값입니다.
SIMPLE	메모리 누수 검출이 활성화되지만, 경고가 발생한 클래스의 이름만 로깅됩니다.
ADVANCED	메모리 누수 검출이 활성화되고, 누수가 발생한 클래스의 스택 트레이스까지 로깅됩니다.

 

 

 

---

 

submit()

비동기적으로 작업을 실행하고, 결과를 Future로 반환합니다.

오버로딩된 여러 버전이 있으며, Callable 또는 Runnable을 인자로 받습니다.

Future<T> submit(Callable<T> task); Future<?> submit(Runnable task); Future<T> submit(Runnable task, T result);

 

 

invokeAll()

여러 작업을 실행하고, 모든 작업의 결과를 Future의 리스트로 반환합니다.

모든 작업이 완료될 때까지 대기합니다.

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;

 

 

invokeAny()

여러 작업을 실행하고, 첫 번째로 완료된 작업의 결과를 반환합니다.

어떤 작업이 완료될 때까지 대기합니다.

<T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;

 

 

shutdown()

더 이상 새로운 작업을 받지 않고, 이미 제출된 작업이 완료될 때까지 대기합니다.

이후에는 스레드 풀을 종료합니다.

 

shutdownNow()

현재 실행 중인 모든 작업을 중단하고, 대기 중인 작업을 취소합니다.

즉시 실행 중인 작업을 중단하려고 시도하고, 남아 있는 작업의 리스트를 반환합니다.

 

isShutdown()

shutdown() 메서드가 호출되었는지 여부를 반환합니다.

 

isTerminated()

모든 작업이 완료되었는지 여부를 반환합니다.

boolean isTerminated();

 

 

awaitTermination()

shutdown() 메서드가 호출된 후, 모든 작업이 완료될 때까지 대기합니다.

최대 대기 시간을 지정할 수 있습니다.

boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

 

 

 

 

 

 

 

 

 

 

---

https://velog.io/@dm911/Java-blocking-IO-%EC%99%80-non-blocking-IO-%EC%97%90-%EB%8C%80%ED%95%B4%EC%84%9C

Java blocking IO 와 non blocking IO 에 대해서..

 

https://blog.naver.com/horajjan/220464906740

[Java] NIO 패키지(java.nio) 소개