EventLoop 내부 동작

// NioEventLoop.run() 핵심 흐름 (단순화)
@Override
protected void run() {
    for (;;) {
        // 1단계: I/O 이벤트 대기
        select(wakenUp.getAndSet(false));

        // 2단계: 발생한 I/O 이벤트 처리
        processSelectedKeys();

        // 3단계: 태스크 큐에 쌓인 작업 실행
        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
    }
}

┌──────────────────────────────────────────────────┐
│                 NioEventLoop.run()                │
│                                                  │
│  ┌────────────┐                                  │
│  │  select()  │◀─────────────── 루프 시작         │
│  │            │                    ▲              │
│  └─────┬──────┘                    │              │
│        │                           │              │
│        ▼                           │              │
│  ┌──────────────────┐              │              │
│  │processSelectedKeys│              │              │
│  │  (I/O 이벤트 처리) │              │              │
│  └─────┬────────────┘              │              │
│        │                           │              │
│        ▼                           │              │
│  ┌──────────────────┐              │              │
│  │  runAllTasks()   │              │              │
│  │  (태스크 큐 실행)  │──────────────┘              │
│  └──────────────────┘                            │
└──────────────────────────────────────────────────┘

// processSelectedKeys 내부 동작 (단순화)
private void processSelectedKeys() {
    Set<SelectionKey> selectedKeys = selector.selectedKeys();
    for (SelectionKey key : selectedKeys) {
        // 각 Key에 attachment로 연결된 NioChannel 가져오기
        NioChannel ch = (NioChannel) key.attachment();

        int readyOps = key.readyOps();

        if ((readyOps & SelectionKey.OP_READ) != 0) {
            // 읽기 가능 → channelRead 이벤트 발생
            ch.unsafe().read();
        }
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            // 쓰기 가능 → flush 처리
            ch.unsafe().forceFlush();
        }
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // 연결 완료 → channelActive 이벤트 발생
            ch.unsafe().finishConnect();
        }
    }
}

// 외부 스레드(비즈니스 로직 스레드 등)에서 EventLoop에 작업 제출
channel.eventLoop().execute(() -> {
    // 이 코드는 EventLoop 스레드에서 실행된다
    channel.writeAndFlush(response);
});

외부 스레드 A ──execute(task1)──▶ ┌──────────┐
                                 │ taskQueue │
외부 스레드 B ──execute(task2)──▶ │ (MPSC)   │──▶ EventLoop 스레드가 꺼내서 실행
                                 │          │
외부 스레드 C ──execute(task3)──▶ └──────────┘

// 별도 스레드풀에서 비즈니스 로직 실행 후, 결과를 Channel로 전송
businessExecutor.submit(() -> {
    MyResponse result = heavyComputation();  // 시간이 오래 걸리는 작업

    // EventLoop 스레드로 돌아가서 write
    channel.eventLoop().execute(() -> {
        channel.writeAndFlush(result);
    });
});

// 5초 후에 한 번 실행
channel.eventLoop().schedule(() -> {
    System.out.println("5초 경과");
}, 5, TimeUnit.SECONDS);

// 1초 간격으로 반복 실행
channel.eventLoop().scheduleAtFixedRate(() -> {
    channel.writeAndFlush(new PingMessage());
}, 0, 1, TimeUnit.SECONDS);

// HashedWheelTimer 사용 예시 — 대량의 연결 타임아웃에 적합
HashedWheelTimer timer = new HashedWheelTimer(
    100, TimeUnit.MILLISECONDS,  // tick 간격
    512                          // 휠 크기
);

// 타이머 콜백은 별도 스레드에서 실행됨
timer.newTimeout(timeout -> {
    // Channel 조작은 EventLoop로 넘겨야 안전
    channel.eventLoop().execute(() -> {
        channel.close();
    });
}, 30, TimeUnit.SECONDS);

// ioRatio 기본값은 50
EventLoopGroup group = new NioEventLoopGroup();

// ioRatio 변경 (NioEventLoop에 직접 접근)
((NioEventLoop) channel.eventLoop()).setIoRatio(70);

ioRatio = 50 (기본값)

┌──────────────────────────────────────────┐
│  select  │  I/O 처리 (10ms)  │ 태스크 (10ms) │
└──────────────────────────────────────────┘

ioRatio = 70

┌──────────────────────────────────────────┐
│  select  │  I/O 처리 (10ms)  │ 태스크 (~4ms)  │
└──────────────────────────────────────────┘

ioRatio = 100 (특수)

┌──────────────────────────────────────────────────┐
│  select  │  I/O 처리  │ 태스크 (시간 제한 없이 전부 실행)  │
└──────────────────────────────────────────────────┘

태스크 허용 시간 = ioTime × (100 - ioRatio) / ioRatio

// EventLoop.execute()의 내부 동작 (단순화)
public void execute(Runnable task) {
    boolean inEventLoop = inEventLoop();  // 현재 스레드가 EventLoop 스레드인가?

    addTask(task);  // 큐에 추가

    if (!inEventLoop) {
        // 외부 스레드 → EventLoop가 select()에서 잠들어 있을 수 있으니 깨운다
        wakeup(inEventLoop);
    }
}

// AbstractChannel.write() 내부 (단순화)
public void write(Object msg) {
    if (eventLoop.inEventLoop()) {
        // EventLoop 스레드에서 호출 → 바로 실행
        doWrite(msg);
    } else {
        // 외부 스레드에서 호출 → execute()로 큐잉
        eventLoop.execute(() -> doWrite(msg));
    }
}

// 절대 하면 안 되는 코드
public class BadHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        Thread.sleep(1000);  // EventLoop 스레드가 1초간 멈춤!
        ctx.writeAndFlush(response);
    }
}

// 절대 하면 안 되는 코드
public class BadDbHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // JDBC 쿼리 — 네트워크 왕복 + 쿼리 실행 시간만큼 블로킹
        ResultSet rs = statement.executeQuery("SELECT * FROM users WHERE id = ?");
        // ...
    }
}

// 절대 하면 안 되는 코드
public class BadHttpHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 동기 HTTP 호출 — 외부 서버 응답 대기 시간만큼 블로킹
        HttpResponse response = httpClient.execute(new HttpGet("https://api.example.com"));
    }
}

// 방법 1: EventExecutorGroup을 파이프라인에 지정
EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(16);

ch.pipeline().addLast(new MyDecoder());
// businessGroup의 스레드에서 실행 — EventLoop를 블로킹하지 않음
ch.pipeline().addLast(businessGroup, new MyBlockingHandler());
ch.pipeline().addLast(new MyEncoder());

// 방법 2: 핸들러 내부에서 직접 별도 스레드풀 사용
public class AsyncDbHandler extends ChannelInboundHandlerAdapter {
    private final ExecutorService dbExecutor = Executors.newFixedThreadPool(16);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // DB 작업을 별도 스레드풀에서 실행
        dbExecutor.submit(() -> {
            String result = queryDatabase(msg);

            // 결과를 EventLoop 스레드로 돌아가서 write
            ctx.channel().eventLoop().execute(() -> {
                ctx.writeAndFlush(result);
            });
        });
    }
}

// EventLoop 스레드에서 이렇게 하면 BlockingOperationException 발생
ChannelFuture future = channel.writeAndFlush(msg);
future.sync();  // 예외! — EventLoop에서 블로킹 대기 시도

// 올바른 방법: 리스너로 비동기 처리
ChannelFuture future = channel.writeAndFlush(msg);
future.addListener((ChannelFutureListener) f -> {
    if (f.isSuccess()) {
        System.out.println("전송 완료");
    } else {
        f.cause().printStackTrace();
    }
});

                         NioEventLoop 전체 동작

   외부 스레드들                           EventLoop 스레드
   ┌────────┐                       ┌──────────────────────────┐
   │Thread-A│──execute(task)──▶     │                          │
   │Thread-B│──schedule(task)──▶    │   for (;;) {             │
   │Thread-C│──execute(task)──▶     │                          │
   └────────┘                       │     ① select()           │
        │                           │        │                 │
        │    wakeup() ─────────────▶│        ▼                 │
        │                           │     ② processSelectedKeys│
        │                           │        (I/O 이벤트 처리)   │
        │                           │        │                 │
        │                           │        ▼                 │
        │                           │     ③ runAllTasks()      │
        └──────▶ taskQueue ────────▶│        (태스크 + 스케줄)   │
                                    │        │                 │
                scheduledTaskQueue──│────────┘                 │
                                    │   }                      │
                                    └──────────────────────────┘