전통적인 블로킹 I/O (BIO)
Thread 1 → Socket 1 (blocking read...)
Thread 2 → Socket 2 (blocking read...)
Thread 3 → Socket 3 (blocking read...)
...
Thread 1000 → Socket 1000 (blocking read...)

문제: 1000개 연결 = 1000개 스레드 = 메모리 부족

Selector 기반 논블로킹 I/O (NIO)
        Selector (1개 스레드)
            ↓
    ┌───────┼───────┐
Channel 1 Channel 2 Channel 3 ... Channel 1000
(wait)    (ready!)   (wait)         (wait)

해결: 1000개 연결 = 1개 스레드 = 메모리 효율적

SelectionKey.OP_ACCEPT   // 서버가 클라이언트 연결 수락 가능
SelectionKey.OP_CONNECT  // 클라이언트가 서버 연결 완료
SelectionKey.OP_READ     // 채널에서 데이터 읽기 가능
SelectionKey.OP_WRITE    // 채널에 데이터 쓰기 가능

전통적인 방식:              select() 방식:
Thread 1 → Socket 1        1개 Thread
Thread 2 → Socket 2            ↓
Thread 3 → Socket 3        select([Socket 1, 2, 3, ...])
...                            ↓
Thread 1000 → Socket 1000  "Socket 2가 준비됨!"
                               ↓
1000개 스레드 필요         1개 스레드로 처리

// Linux select() system call
int select(int nfds,
           fd_set *readfds,    // 읽기 가능한 파일 디스크립터
           fd_set *writefds,   // 쓰기 가능한 파일 디스크립터
           fd_set *exceptfds,  // 예외 발생한 파일 디스크립터
           struct timeval *timeout);

1. User Space → Kernel Space
   fd_set 전체를 복사 (1000개 등록되어 있으면 1000개 전부)

2. Kernel에서 처리
   각 fd를 순회하며 이벤트 발생 여부 확인

3. Kernel Space → User Space
   결과를 다시 복사 (준비된 fd 개수와 무관하게 전체 fd_set)

// Linux poll() system call
int poll(struct pollfd *fds,  // 파일 디스크립터 배열
         nfds_t nfds,          // 배열 크기
         int timeout);

struct pollfd {
    int fd;           // 파일 디스크립터
    short events;     // 관심 이벤트
    short revents;    // 실제 발생한 이벤트
};

// epoll 생성
int epfd = epoll_create1(0);

// fd 등록 (한 번만!)
struct epoll_event event;
event.events = EPOLLIN;  // 읽기 이벤트
event.data.fd = sockfd;
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &event);

// 이벤트 대기
struct epoll_event events[MAX_EVENTS];
int nfds = epoll_wait(epfd, events, MAX_EVENTS, timeout);

// 준비된 fd만 반환됨! (O(1) 조회)
for (int i = 0; i < nfds; i++) {
    if (events[i].events & EPOLLIN) {
        // 읽기 가능한 소켓 처리
        int fd = events[i].data.fd;
        read(fd, buffer, size);
    }
}

int epfd = epoll_create1(0);

struct epoll_event event;
event.events = EPOLLIN;  // 관심 이벤트
event.data.fd = sockfd;

// 등록
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &event);

// 수정
epoll_ctl(epfd, EPOLL_CTL_MOD, sockfd, &event);

// 삭제
epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);

struct epoll_event events[MAX_EVENTS];
int nfds = epoll_wait(epfd, events, MAX_EVENTS, timeout);

// timeout 예시
epoll_wait(epfd, events, MAX_EVENTS, -1);     // 무한 대기 (이벤트 올 때까지)
epoll_wait(epfd, events, MAX_EVENTS, 0);      // 즉시 반환 (논블로킹)
epoll_wait(epfd, events, MAX_EVENTS, 5000);   // 5초 대기

[이벤트 대기]
epoll_wait() → 블로킹 (이벤트 올 때까지 대기)
    ↓
패킷 도착!
    ↓
epoll_wait() 반환

[I/O 작업]
read(fd, buffer, size) → 논블로킹 (데이터가 이미 준비됨)
    ↓
즉시 읽기 완료

// 1. Selector.open() → epoll_create1()
Selector selector = Selector.open();
//   ↓ JNI
// int epfd = epoll_create1(0);

// 2. channel.register() → epoll_ctl(EPOLL_CTL_ADD)
ServerSocketChannel serverSocket = ServerSocketChannel.open();
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
//   ↓ JNI
// struct epoll_event event;
// event.events = EPOLLIN;
// epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event);

// 3. selector.select() → epoll_wait()
int readyChannels = selector.select();  // 블로킹!
//   ↓ JNI
// int nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);

// 4. selector.select(timeout) → epoll_wait(timeout)
int readyChannels = selector.select(5000);  // 5초 대기
//   ↓ JNI
// int nfds = epoll_wait(epfd, events, MAX_EVENTS, 5000);

// 5. selector.selectNow() → epoll_wait(0)
int readyChannels = selector.selectNow();  // 즉시 반환
//   ↓ JNI
// int nfds = epoll_wait(epfd, events, MAX_EVENTS, 0);

1. Selector.open()
   → epoll_create1()
   → 커널에 epoll 인스턴스 생성

2. channel.register(selector, ops)
   → epoll_ctl(EPOLL_CTL_ADD)
   → Red-Black Tree에 fd 등록

3. selector.select()
   → epoll_wait()
   → Ready List 확인 (블로킹 대기)

// 전통적인 블로킹 I/O
ServerSocket serverSocket = new ServerSocket(8080);

while (true) {
    Socket socket = serverSocket.accept();  // 블로킹 1: 연결 대기

    new Thread(() -> {
        try {
            InputStream in = socket.getInputStream();
            byte[] buffer = new byte[1024];

            while (true) {
                int bytesRead = in.read(buffer);  // 블로킹 2: 데이터 대기
                if (bytesRead == -1) break;

                // 데이터 처리
                socket.getOutputStream().write(buffer, 0, bytesRead);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }).start();
}

// 문제점:
// - 1000개 연결 = 1000개 스레드
// - 각 스레드가 read()에서 블로킹
// - 메모리 낭비 (스레드당 1MB 스택)

// Acceptor 스레드 (1개): 연결 수락 전담
class Acceptor implements Runnable {
    public void run() {
        while (running) {
            SocketChannel socket = serverSocket.accept();  // 블로킹: 연결 대기
            socket.configureBlocking(false);

            // Poller에게 소켓 등록 요청
            poller.register(socket);
        }
    }
}

// Poller 스레드 (1-2개): 이벤트 감지 전담
class Poller implements Runnable {
    private Selector selector = Selector.open();

    public void run() {
        while (running) {
            selector.select();  // 블로킹: 이벤트 대기

            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isReadable()) {
                    // Worker 스레드 풀에 작업 위임
                    workerExecutor.execute(new SocketProcessor(key));
                }
            }
        }
    }

    public void register(SocketChannel socket) {
        socket.register(selector, SelectionKey.OP_READ);
    }
}

// Worker 스레드 (200개 기본값): 실제 I/O 및 비즈니스 로직 처리
class SocketProcessor implements Runnable {
    public void run() {
        ByteBuffer buffer = ByteBuffer.allocate(8192);
        int bytesRead = socket.read(buffer);  // 논블로킹: 즉시 읽기

        // HTTP 파싱, Servlet 실행, 응답 생성
        processRequest(buffer);

        socket.write(responseBuffer);  // 논블로킹: 즉시 쓰기
    }
}

// 장점:
// - 역할 분리: Acceptor(연결) / Poller(이벤트) / Worker(처리)
// - 1000개 연결 = 3개 스레드 (Acceptor 1 + Poller 1-2 + Worker 필요시)
// - Poller는 이벤트만 감지, 실제 작업은 Worker 풀 사용
// - 메모리 효율적

[전통적인 블로킹 I/O]
accept() → 블로킹 (연결 대기)
read() → 블로킹 (데이터 대기)
write() → 블로킹 (버퍼 공간 대기)

→ 스레드가 I/O 작업마다 블로킹됨
→ 1000개 연결 = 1000개 스레드 필요

[Tomcat NIO - 이벤트 기반 + 스레드 분리]
Acceptor: serverSocket.accept() → 블로킹 (연결 대기)
Poller: selector.select() → 블로킹 (이벤트 대기)
Worker: read() → 논블로킹 (데이터 준비됨)
Worker: write() → 논블로킹 (버퍼 준비됨)

→ Acceptor/Poller는 블로킹, 실제 I/O는 논블로킹
→ 1000개 연결 = 3개 스레드 (Acceptor 1 + Poller 1-2 + Worker 필요시)

// kqueue 생성
int kq = kqueue();

// 이벤트 등록
struct kevent change;
EV_SET(&change, sockfd, EVFILT_READ, EV_ADD, 0, 0, NULL);
kevent(kq, &change, 1, NULL, 0, NULL);

// 이벤트 대기
struct kevent event;
int nev = kevent(kq, NULL, 0, &event, 1, NULL);

if (nev > 0 && event.filter == EVFILT_READ) {
    // 읽기 가능
    read(event.ident, buffer, size);
}

      Java Selector.select()
                ↓
    JNI (Java Native Interface)
                ↓
┌──────────────────────────────────┐
│   Linux    │   macOS   │ Windows │
│   epoll()  │  kqueue() │  IOCP   │
└──────────────────────────────────┘
                ↓
      커널이 파일 디스크립터 모니터링
                ↓
       네트워크 카드에서 패킷 도착
                ↓
          커널 인터럽트 발생
                ↓
        Ready List에 fd 추가
                ↓
         epoll_wait() 반환
                ↓
      Java Selector.select() 반환

EventLoopGroup (8개 스레드)
├── EventLoop-1 (스레드 1)
│   ├── Channel A → 모든 이벤트 항상 여기서 처리
│   └── Channel B
│
├── EventLoop-2 (스레드 2)
│   └── Channel C
│
├── EventLoop-3 (스레드 3)
    ├── Channel D
    ├── Channel E
    └── Channel F

Tomcat: 요청마다 Worker 풀에서 임의 할당
Request 1 → Worker-5
Request 2 → Worker-3 (같은 연결인데 다른 스레드)
Request 3 → Worker-7

Netty: 채널당 고정 EventLoop
Channel A의 모든 이벤트 → EventLoop-1 (항상 동일)

// Tomcat: 기본 Java NIO Selector
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();

// Netty: Selector 내부를 Array로 교체
for (int i = 0; i < selectedKeys.size; ++i) {
    SelectionKey key = selectedKeys.keys[i];
}
// Array는 연속된 메모리 공간 → CPU 캐시 지역성 향상

worker.execute(() -> {
    // HTTP 파싱 (논블로킹)
    HttpServletRequest request = parseRequest(socket);

    // Servlet 호출 (블로킹 가능)
    InputStream in = request.getInputStream();
    byte[] body = in.readAllBytes();  // 전체 바디를 다 받을 때까지 블로킹!

    // 10MB 바디가 천천히 도착하면?
    // → Worker 스레드가 계속 대기
    // → 다른 요청 처리 불가
});

channel.pipeline()
    .addLast(new HttpServerCodec())        // 논블로킹 디코딩
    .addLast(new MyHandler() {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            // 소켓에 준비된 만큼 읽음 (가변적: 100바이트 ~ 수KB)
            // → channelRead() 호출
            // 또 준비된 만큼 읽음 → channelRead() 다시 호출
            // ...반복...

            // HttpServerCodec이 조각들을 모아서
            // 완전한 HttpRequest로 만들어서 전달
            HttpRequest request = (HttpRequest) msg;

            // EventLoop 스레드는 즉시 리턴
            // 다른 채널의 이벤트 처리 가능
        }
    });

[Tomcat]
10MB 바디 천천히 도착
→ Worker 스레드가 전체 받을 때까지 블로킹
→ 해당 스레드는 다른 요청 처리 불가

[Netty]
10MB 바디 천천히 도착
→ 준비된 만큼 읽을 때마다 channelRead() 호출 (가변적)
→ EventLoop 스레드는 즉시 반환
→ 다른 수천 개 채널 처리 가능

// 개발자가 작성
channel.pipeline()
    .addLast(new HttpServerCodec())      // Netty 제공
    .addLast(new MyHandler(gptClient, dbExecutor)); // 개발자 구현

public class MyHandler extends ChannelInboundHandlerAdapter {
    private final GptClient gptClient;
    private final ExecutorService dbExecutor;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        HttpRequest request = (HttpRequest) msg;

        // GPT API 호출 (논블로킹)
        gptClient.callAsync(request)
            .thenCompose(gptResponse ->
                // GPT 응답을 DB에 저장 (블로킹, 별도 스레드 풀)
                CompletableFuture.supplyAsync(() ->
                    chatHistoryRepository.save(new ChatHistory(gptResponse)), dbExecutor
                ).thenApply(savedHistory -> gptResponse)
            )
            .thenAccept(gptResponse -> {
                HttpResponse response = createResponse(gptResponse);
                ctx.writeAndFlush(response);  // EventLoop의 taskQueue에 제출
            });

        // 즉시 리턴! EventLoop 스레드는 다른 채널 처리 가능
    }
}

1. EventLoop 스레드: channelRead() 호출
2. EventLoop 스레드: GPT API 비동기 호출하고 즉시 리턴
3. EventLoop 스레드: GPT 응답 수신 (논블로킹)
4. dbExecutor 스레드: DB 저장 (블로킹)
5. dbExecutor 스레드: thenAccept() 콜백 실행
6. dbExecutor 스레드: ctx.writeAndFlush() → EventLoop에 Task 제출
7. EventLoop 스레드: Task 실행하여 네트워크 전송

// ❌ 잘못된 코드: 동기 블로킹 작업
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    HttpRequest request = (HttpRequest) msg;

    // EventLoop 스레드가 GPT 응답 올 때까지 블로킹!
    GptResponse gptResponse = gptClient.callSync(request);  // 블로킹!

    // DB 저장까지 블로킹!
    chatHistoryRepository.save(new ChatHistory(gptResponse));  // 블로킹!

    // 다른 수천 개 채널도 같이 대기
    // Netty의 장점 완전히 상실!

    HttpResponse response = createResponse(gptResponse);
    ctx.writeAndFlush(response);
}

// 예: GPT 응답 받고 DB 저장 후 응답 전송
gptClient.callAsync(request)
    .thenCompose(gptResponse ->
        CompletableFuture.supplyAsync(() ->
            chatHistoryRepository.save(new ChatHistory(gptResponse)), dbExecutor
        ).thenApply(savedHistory -> gptResponse)
    )
    .thenAccept(gptResponse -> {
        HttpResponse response = createResponse(gptResponse);
        ctx.writeAndFlush(response);  // ← dbExecutor 스레드가 호출
    });

// Tomcat: Worker 스레드가 모든 작업을 순차 처리
read() → 파싱 → 비즈니스 로직 → 응답 생성 → write()
// I/O와 비즈니스 로직이 분리되지 않음

// Netty: I/O와 Task를 비율에 따라 처리
long ioStartTime = System.nanoTime();
processSelectedKeys();  // I/O 이벤트 처리 (읽기/쓰기)

long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);  // Task 처리
// I/O:Task = 50:50 (ioRatio 기본값)

// Netty: JDK epoll bug 자동 감지 및 복구
if (unexpectedSelectorWakeup(selectCnt)) {
    rebuildSelector();  // Selector 재구성
    selectCnt = 0;
}

// Tomcat: 기본 JDK Selector에 의존

// Task가 없을 때: I/O 이벤트 대기
if (!hasTasks()) {
    selector.select(timeout);  // 블로킹: I/O 이벤트 올 때까지 대기
    processSelectedKeys();     // I/O 처리
}

// Task가 있을 때: 블로킹 스킵
if (hasTasks()) {
    selector.selectNow();      // 논블로킹: 즉시 확인하고 리턴
    processSelectedKeys();     // I/O 이벤트 있으면 처리
    runAllTasks();             // Task 처리
}

// 잘못된 예시: select() 없이 Task만 처리
while (hasTasks()) {
    runAllTasks();  // Task만 계속 처리
}
// 이 동안 클라이언트가 보낸 요청은?
// → 읽지 못함! 클라이언트는 응답 못 받고 타임아웃

// 올바른 예시: selectNow()로 I/O도 확인
if (hasTasks()) {
    selector.selectNow();      // I/O 이벤트 빠르게 확인
    processSelectedKeys();     // 있으면 처리
    runAllTasks();             // Task 처리
}

// 1. Pipeline 생성 및 Handler 추가
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast("codec", new HttpClientCodec());           // Handler 1
pipeline.addLast("aggregator", new HttpObjectAggregator()); // Handler 2
pipeline.addLast("reactor", new ReactorNettyHandler());     // Handler 3

// 2. 이벤트 발생 시 Pipeline이 Handler들을 순차 실행
pipeline.fireChannelRead(buffer);
//   ↓
// HttpClientCodec.channelRead() 실행
//   ↓
// HttpObjectAggregator.channelRead() 실행
//   ↓
// ReactorNettyHandler.channelRead() 실행

// Channel 생성 시 Bootstrap에서 Handler 등록
Bootstrap b = new Bootstrap();
b.handler(new ChannelInitializer<>() {
    @Override
    protected void initChannel(Channel ch) {
        ChannelPipeline p = ch.pipeline();

        // Handler들을 Pipeline에 순서대로 추가
        p.addLast(new HttpClientCodec());
        p.addLast(new HttpObjectAggregator(8192));
        p.addLast(new ReactorNettyHandler());
    }
});

// 데이터 읽을 때
socketChannel.read(buffer);  // OS에서 데이터 읽기
pipeline.fireChannelRead(buffer);  // Pipeline의 모든 Handler 순차 실행

// Spring WebClient 내부에서 자동으로 실행
channel.pipeline()
    .addLast(new HttpClientCodec())           // Spring이 자동 추가
    .addLast(new HttpObjectAggregator(256KB)) // Spring이 자동 추가
    .addLast(new ReactorNettyHandler());      // Spring이 자동 추가

서버가 전송:
Chunk 1: "Hello "
Chunk 2: "World"
Chunk 3: "!"

HttpObjectAggregator가 병합
→ 최종 결과: "Hello World!"

// 인코딩 (요청 전송)
User user = new User("John");
→ Codec: Object를 JSON 바이트로 변환
→ {"name":"John"}
→ 네트워크 전송

// 디코딩 (응답 수신)
네트워크에서 바이트 수신
→ {"name":"John"}
→ Codec: JSON 바이트를 User 객체로 변환
→ User user = ...

@Bean
public WebClient webClient() {
    return WebClient.builder()
        .codecs(configurer -> configurer.defaultCodecs()
            .maxInMemorySize(16 * 1024 * 1024))  // 16MB 제한
        .build();
}

실제 HTTP 통신을 할 수 있는 라이브러리들:
- Reactor Netty (논블로킹)
- Jetty (논블로킹)
- Apache HttpClient (블로킹/논블로킹)

WebClient (Spring 표준 인터페이스)
    ↓
ReactorClientHttpConnector (어댑터)
    ↓
HttpClient (Reactor Netty)
    ↓
NioEventLoop (실제 네트워크)

@Bean
public WebClient webClient() {
    // 1. Reactor Netty 사용
    HttpClient nettyClient = HttpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
        .responseTimeout(Duration.ofSeconds(10));

    return WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(nettyClient))
        .build();

    // 2. Jetty 사용 가능
    // .clientConnector(new JettyClientHttpConnector())

    // 3. Apache HttpClient 사용 가능
    // .clientConnector(new HttpComponentsClientHttpConnector())
}

// WebClient 내부 코드 (단순화)
public class WebClient {
    private ClientHttpConnector connector;

    public Mono<String> get() {
        // WebClient는 HTTP 통신 방법을 모름
        // 커넥터에게 위임
        return connector.connect(request)
            .flatMap(connection -> ...);
    }
}

// ReactorClientHttpConnector 내부
public class ReactorClientHttpConnector implements ClientHttpConnector {
    private HttpClient httpClient;  // Netty

    @Override
    public Mono<Connection> connect(HttpRequest request) {
        // Netty에게 실제 HTTP 통신 요청
        return httpClient.connect();
    }
}

// Spring Controller (http-nio-8080-exec-5 스레드)
@PostMapping("/chat")
public Mono<ChatResponse> chat(@RequestBody ChatRequest request) {
    // GPT API 호출
    return webClient.post()
        .uri("https://api.openai.com/v1/chat/completions")
        .bodyValue(createGptRequest(request.message()))
        .retrieve()
        .bodyToMono(GptResponse.class)
        .flatMap(gptResponse ->
            // GPT 응답을 DB에 저장 (블로킹 작업)
            Mono.fromCallable(() ->
                chatHistoryRepository.save(new ChatHistory(request, gptResponse))
            ).subscribeOn(Schedulers.boundedElastic())
            .thenReturn(new ChatResponse(gptResponse))
        );
    // Mono 반환 후 즉시 리턴 → Tomcat 스레드는 다른 요청 처리 가능
}

// DefaultWebClient.java
private Mono<ClientResponse> exchange() {
    return Mono.defer(() -> {
        // HTTP 요청 생성
        ClientHttpRequest httpRequest = createRequest();

        // ReactorClientHttpConnector로 전달
        return this.connector.connect(
            this.method,
            this.uri,
            requestCallback
        );
    });
}

// ReactorClientHttpConnector.java
@Override
public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri, ...) {
    return this.httpClient
        .request(adaptHttpMethod(method))
        .uri(uri.toString())
        .send((request, outbound) -> {
            // 요청 본문 전송
            return requestCallback.apply(new ReactorClientHttpRequest(...));
        })
        .responseConnection((response, connection) -> {
            // 응답 수신
            return Mono.just(new ReactorClientHttpResponse(...));
        });
}

// Bootstrap.java
public ChannelFuture connect(SocketAddress remoteAddress) {
    ChannelFuture regFuture = initAndRegister();

    // Channel을 EventLoop에 등록
    Channel channel = regFuture.channel();
    EventLoop eventLoop = channel.eventLoop();

    // EventLoop에서 실행
    eventLoop.execute(() -> {
        channel.register(selector, SelectionKey.OP_CONNECT);
        channel.connect(remoteAddress);
    });

    return regFuture;
}

// NioEventLoop.run() - reactor-http-nio-2 스레드
for (;;) {
    // 1. I/O 이벤트 대기
    selector.select(timeoutMillis);
    //   ↓
    // JNI → epoll_wait()
    //   ↓
    // 커널이 Ready List 확인
    //   ↓
    // HTTP 응답 패킷 도착!
    //   ↓
    // fd가 READ 가능 상태가 됨
    //   ↓
    // epoll_wait() 반환

    // 2. 준비된 채널 처리
    processSelectedKeys();
    //   ↓
    for (int i = 0; i < selectedKeys.size; i++) {
        SelectionKey key = selectedKeys.keys[i];
        AbstractNioChannel channel = (AbstractNioChannel) key.attachment();

        if (key.readyOps & OP_READ) {
            // 3. 데이터 읽기
            channel.unsafe().read();
            //   ↓
            // ByteBuffer에 데이터 읽기
            ByteBuffer buffer = ByteBuffer.allocate(8192);
            int bytesRead = socketChannel.read(buffer);
            //   ↓
            // 4. Pipeline 처리
            ChannelPipeline pipeline = channel.pipeline();
            pipeline.fireChannelRead(buffer);
            //   ↓
            // HttpClientCodec: HTTP 파싱
            // HttpObjectAggregator: 청크 병합
            // ReactorNettyHandler: Mono/Flux로 변환
            //   ↓
            // 5. Mono에 데이터 전달
            sink.next(httpResponse);
        }
    }

    // 6. 일반 Task 처리
    runAllTasks();
}

// reactor-http-nio-2 스레드에서 GPT 응답 수신
mono
    .doOnNext(gptResponse -> {
        // EventLoop 스레드에서 실행!
        log.info("Thread: {}", Thread.currentThread().getName());
        // 출력: Thread: reactor-http-nio-2
        log.info("GPT Response: {}", gptResponse.getChoices().get(0).getMessage());
    })
    .flatMap(gptResponse ->
        // DB 저장은 boundedElastic 스레드로 위임
        Mono.fromCallable(() ->
            chatHistoryRepository.save(new ChatHistory(gptResponse))
        ).subscribeOn(Schedulers.boundedElastic())
        .thenReturn(gptResponse)
    )
    .subscribe(finalResponse -> {
        // 사용자에게 응답 전송
        log.info("최종 응답 전송");
    });

// 잘못된 예시: EventLoop 스레드에서 블로킹
@PostMapping("/chat")
public Mono<ChatResponse> chat(@RequestBody ChatRequest request) {
    return webClient.post()
        .uri("https://api.openai.com/v1/chat/completions")
        .bodyValue(createGptRequest(request.message()))
        .retrieve()
        .bodyToMono(GptResponse.class)
        .map(gptResponse -> {
            // ❌ EventLoop 스레드에서 블로킹 DB 작업!
            chatHistoryRepository.save(new ChatHistory(gptResponse));
            return new ChatResponse(gptResponse);
        });
    // subscribeOn() 없음 → reactor-http-nio 스레드에서 실행
}

EventLoop 스레드 (reactor-http-nio-2)
    ↓
GPT 응답 수신 (논블로킹 ✓)
    ↓
DB 저장 시작 (블로킹!)
    ↓
DB 응답 올 때까지 대기... (100ms)
    ↓
이 EventLoop가 담당하는 수천 개 채널도 모두 대기!
    ↓
Netty의 장점 완전히 상실

// 올바른 코드: Bounded 스레드에서 블로킹
@PostMapping("/chat")
public Mono<ChatResponse> chat(@RequestBody ChatRequest request) {
    return webClient.post()
        .uri("https://api.openai.com/v1/chat/completions")
        .bodyValue(createGptRequest(request.message()))
        .retrieve()
        .bodyToMono(GptResponse.class)
        .flatMap(gptResponse ->
            // ✓ boundedElastic 스레드에서 DB 저장
            Mono.fromCallable(() ->
                chatHistoryRepository.save(new ChatHistory(gptResponse))
            ).subscribeOn(Schedulers.boundedElastic())
            .thenReturn(new ChatResponse(gptResponse))
        );
}

EventLoop 스레드 (reactor-http-nio-2)
    ↓
GPT 응답 수신 (논블로킹)
    ↓
Mono 생성하고 즉시 리턴 (논블로킹!)
    ↓
다른 채널 처리 가능

별도로:
boundedElastic-1 스레드
    ↓
DB 저장 (블로킹)
    ↓
완료 후 EventLoop에 응답 전달

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    HttpRequest request = (HttpRequest) msg;

    // GPT API 호출 후 DB 저장
    gptClient.callAsync(request)
        .thenCompose(gptResponse ->
            CompletableFuture.supplyAsync(() ->
                chatHistoryRepository.save(new ChatHistory(gptResponse)), dbExecutor
            ).thenApply(savedHistory -> gptResponse)
        )
        .thenAccept(gptResponse -> {
            HttpResponse response = createResponse(gptResponse);
            ctx.writeAndFlush(response);  // ← 개발자가 직접!
            // → EventLoop taskQueue에 제출
        });
}

@PostMapping("/chat")
public Mono<ChatResponse> chat(@RequestBody ChatRequest request) {
    return webClient.post()
        .uri("https://api.openai.com/v1/chat/completions")
        .bodyValue(createGptRequest(request.message()))
        .retrieve()
        .bodyToMono(GptResponse.class)
        .flatMap(gptResponse ->
            Mono.fromCallable(() ->
                chatHistoryRepository.save(new ChatHistory(gptResponse))
            ).subscribeOn(Schedulers.boundedElastic())
            .thenReturn(new ChatResponse(gptResponse))
        );

    // ← 그냥 Mono 반환만!
    // Spring이 알아서 ctx.writeAndFlush() 해줌
}

// Spring WebFlux 내부 코드 (단순화)
mono.subscribe(result -> {
    // 1. HTTP 응답 생성
    HttpResponse response = createHttpResponse(result);

    // 2. 자동으로 ctx.writeAndFlush() 호출!
    ctx.writeAndFlush(response);
    // → EventLoop taskQueue에 자동 제출
});

1. Tomcat Worker 스레드: 사용자로부터 HTTP 요청 수신
2. Tomcat Worker 스레드: WebClient로 GPT 서버 요청, Mono 생성하고 즉시 리턴
3. EventLoop 스레드 (reactor-http-nio): GPT 서버에 HTTP 요청 전송 (논블로킹)
4. EventLoop 스레드: GPT 서버 응답 수신 (논블로킹)
5. EventLoop 스레드: Mono에 응답 데이터 전달
6. boundedElastic 스레드: 응답 데이터 블로킹 처리 (DB 저장 등)
7. boundedElastic 스레드: 처리 완료, 최종 응답 생성
8. Tomcat Worker 스레드: 사용자에게 최종 응답 전송

// Tomcat Poller 스레드
ByteBuffer heapBuffer = ByteBuffer.allocate(8192);  // JVM 힙 메모리
socketChannel.read(heapBuffer);  // 커널 버퍼 → JVM 힙

// Spring Controller에서
@PostMapping("/api")
public User create(@RequestBody User user) {
    // Spring 내부에서 수행되는 작업:
    // 1. request.getInputStream() 호출
    // 2. HeapByteBuffer → byte[] 배열로 복사
    // 3. Jackson이 byte[]를 User 객체로 역직렬화
    return user;
}

네트워크 카드
    ↓ DMA 복사 (하드웨어)
커널 버퍼 (OS)
    ↓ CPU 복사 (1) - socketChannel.read()
HeapByteBuffer (JVM 힙 메모리)
    ↓ CPU 복사 (2) - request.getInputStream().read()
byte[] 배열 (JVM 힙 메모리)
    ↓ CPU 복사 (3) - Jackson 역직렬화
Java Object (User, DTO 등)

총 3번의 CPU 복사 발생!

// Netty EventLoop 스레드
ByteBuf buffer = channel.alloc().directBuffer(8192);  // Native 메모리
channel.read(buffer);  // 커널 버퍼 → DirectBuffer

// Handler에서 바로 사용
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ByteBuf data = (ByteBuf) msg;
    // 바로 처리 가능 (추가 복사 없음)
}

네트워크 카드
    ↓ DMA 복사
커널 버퍼
    ↓ CPU 복사 (1) - JVM 힙 우회!
DirectBuffer (Native 메모리)
    ↓
애플리케이션 접근

총 1번의 CPU 복사 발생 (JVM 힙 우회로 GC 압박 없음)

// Netty Handler
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ByteBuf directBuf = (ByteBuf) msg;  // DirectBuffer

    // HTTP 헤더 파싱 - DirectBuffer에서 직접 읽음 (복사 없음!)
    byte method = directBuf.readByte();  // 'G' (GET)
    byte[] uriBytes = new byte[10];
    directBuf.readBytes(uriBytes);       // "/api/users"

    // 헤더 정보 추출 - Native 메모리에서 직접 읽음
    String contentType = parseHeader(directBuf, "Content-Type");
    int contentLength = parseInt(parseHeader(directBuf, "Content-Length"));

    // → 복사 없이 DirectBuffer에서 바로 파싱
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    HttpRequest request = (HttpRequest) msg;
    ByteBuf bodyBuf = ((HttpContent) msg).content();  // DirectBuffer

    // 1. 단순 전달 (파일 다운로드, 프록시 등): 복사 없음
    if (isProxyRequest()) {
        ctx.write(bodyBuf);  // DirectBuffer → 소켓 (복사 없음!)
        return;
    }

    // 2. 비즈니스 로직 필요 (JSON → Java 객체): 힙으로 복사
    if (needsDeserialization()) {
        byte[] bodyBytes = new byte[bodyBuf.readableBytes()];
        bodyBuf.readBytes(bodyBytes);  // DirectBuffer → byte[] (복사 발생!)

        User user = objectMapper.readValue(bodyBytes, User.class);
        // → Java 객체로 변환하려면 힙 메모리 필요
    }
}

커널 버퍼
    ↓ CPU 복사 (1)
DirectBuffer (Native 메모리)
    ↓
Header 파싱 (복사 없음)
    ↓
라우팅, 인증 체크 등

총 복사: 1번

커널 버퍼
    ↓ CPU 복사 (1)
DirectBuffer (Native 메모리)
    ↓
바로 소켓으로 전송 (복사 없음)

총 복사: 1번

커널 버퍼
    ↓ CPU 복사 (1)
DirectBuffer (Native 메모리)
    ↓ CPU 복사 (2)
byte[] (JVM 힙)
    ↓ CPU 복사 (3) - Jackson 내부 복사
Java Object (User, DTO 등)

총 복사: 3번

커널 버퍼 → JVM 힙 → byte[] → Object
(헤더/바디 구분 없이 무조건 3번 CPU 복사)

커널 버퍼 → Native 메모리 (1번 복사)
    ↓
헤더: DirectBuffer에서 직접 파싱 → 총 1번 복사
바디 (프록시/파일): DirectBuffer → 소켓 → 총 1번 복사
바디 (비즈니스 로직): DirectBuffer → byte[] → Object → 총 3번 복사

// 여러 버퍼를 복사 없이 결합
ByteBuf header = ...;
ByteBuf body = ...;

CompositeByteBuf composite = Unpooled.compositeBuffer();
composite.addComponents(true, header, body);
// 실제 복사 없음! 논리적 결합만

[Tomcat]
헤더 버퍼 + 바디 버퍼
    ↓
새 버퍼 생성 후 물리적 복사 (수 MB 복사!)
    ↓
소켓으로 전송

[Netty - CompositeByteBuf]
헤더 버퍼 + 바디 버퍼
    ↓
논리적 결합 (포인터 2개만 저장, 16바이트)
    ↓
두 버퍼를 순차적으로 소켓 전송

class CompositeByteBuf {
    private Component[] components;  // 버퍼 참조 배열

    static class Component {
        ByteBuf buf;           // 실제 버퍼 (복사 안 함)
        int offset;            // 논리적 시작 위치
        int endOffset;         // 논리적 끝 위치
    }
}

// CompositeByteBuf에서 데이터 읽기
composite.readByte();

// 내부적으로
int componentIndex = findComponent(readerIndex);
Component c = components[componentIndex];
return c.buf.getByte(readerIndex - c.offset);
// → 실제 버퍼에서 직접 읽음 (복사 없음)

// HTTP 응답 생성 시
ByteBuf header = Unpooled.copiedBuffer("HTTP/1.1 200 OK\r\n...", UTF_8);
ByteBuf body = readFromFile();  // 파일에서 읽은 큰 데이터

// 전통적 방식: 새 버퍼 생성 후 복사
ByteBuf response = Unpooled.buffer(header.readableBytes() + body.readableBytes());
response.writeBytes(header);  // 복사 1
response.writeBytes(body);    // 복사 2 (큰 데이터!)

// CompositeByteBuf: 복사 없이 결합
CompositeByteBuf composite = Unpooled.compositeBuffer();
composite.addComponents(true, header, body);  // 참조만 저장
channel.write(composite);  // 두 버퍼를 순차적으로 전송

class DirectByteBuffer extends ByteBuffer {
    private long address;      // Native 메모리 주소 (포인터)
    private Cleaner cleaner;   // Native 메모리 해제용
    // ... 기타 필드들 (~100바이트)
}

JVM 힙 메모리
┌─────────────────────────────┐
│ DirectByteBuffer 객체        │ ← GC 대상!
│ - address: 0x7f8a3c000000   │ (객체 크기: ~100바이트)
│ - cleaner: Cleaner 객체      │
└─────────────────────────────┘
         │
         │ 포인터
         ↓
Native 메모리 (힙 외부)
┌─────────────────────────────┐
│ 실제 데이터 버퍼 (8KB)          │ ← GC가 직접 수거 불가!
│ [바이트 데이터...]             │   (malloc으로 할당)
└─────────────────────────────┘

1. DirectByteBuffer 객체가 더 이상 참조되지 않음
   ↓
2. GC가 DirectByteBuffer 객체(힙)를 수거 대상으로 식별
   ↓
3. Cleaner.clean() 실행 (finalization)
   ↓
4. Cleaner가 free(address) 호출
   ↓
5. Native 메모리(힙 외부) 해제

Eden 영역에 생성
    ↓
계속 참조됨 (여러 요청 처리)
    ↓
Minor GC 살아남음
    ↓
Old Generation으로 승격
    ↓
Minor GC로는 수거 불가 
    ↓
Major/Full GC 필요 (느리고 자주 안 일어남)

// 매 요청마다 새 버퍼 생성
for (int i = 0; i < 100000; i++) {
    ByteBuf buffer = Unpooled.directBuffer(8192);  // Native 메모리 할당 (비용 큼!)
    buffer.writeBytes(data);
    channel.write(buffer);
    // GC가 나중에 수거 → GC 압박 증가
}

// Netty의 PooledByteBufAllocator
ByteBufAllocator alloc = PooledByteBufAllocator.DEFAULT;

// 풀에서 재사용
ByteBuf buffer = alloc.buffer(8192);  // 풀에서 빌림 (생성 X)
try {
    buffer.writeBytes(data);
    channel.write(buffer);
} finally {
    buffer.release();  // 풀로 반환 (GC 대상 아님!)
}

[버퍼 풀]
┌─────────────────────────────┐
│ ByteBuf[8KB] - 사용 가능      │
│ ByteBuf[8KB] - 사용 중       │
│ ByteBuf[8KB] - 사용 가능      │
│ ByteBuf[16KB] - 사용 가능     │
└─────────────────────────────┘

요청 1: alloc.buffer(8192)
    → 풀에서 빌림 (생성 X)
    → 사용
    → buffer.release()
    → 풀로 반환

요청 2: alloc.buffer(8192)
    → 방금 반환된 버퍼 재사용!

[매번 새로 생성 (일반적인 방식)]
매 요청:
new DirectByteBuffer(8192)
    ↓ JVM 힙: 객체 생성 (~100바이트)
    ↓ Native 메모리: malloc() 호출 (8KB) → 100us
   사용
    ↓
GC가 DirectByteBuffer 객체(힙) 수거
    ↓
Cleaner가 Native 메모리 해제 (free 호출)
    ↓
GC 일시 정지 발생

[Netty Pooling]
매 요청:
풀에서 빌림                 // 배열 접근 → 1us
    ↓
   사용
    ↓
 풀로 반환                  // 배열 삽입 → 1us
    ↓
GC 대상 아님 (계속 재사용)

[개별 처리 - 비효율적]
write(buffer1)  → 시스템 콜 (User → Kernel 전환: ~100ns)
write(buffer2)  → 시스템 콜 (~100ns)
write(buffer3)  → 시스템 콜 (~100ns)
총 시스템 콜: 3번 (300ns)

[배치 처리 - 효율적]
write([buffer1, buffer2, buffer3])  → 시스템 콜 1번 (~100ns)
총 시스템 콜: 1번 (100ns)

→ 시스템 콜 횟수 67% 감소!

// 여러 write 호출
channel.write(header);   // 즉시 전송 안 함! 버퍼에 저장
channel.write(body);     // 버퍼에 저장
channel.write(footer);   // 버퍼에 저장

// flush 호출 시 한 번에 전송
channel.flush();  // 3개를 모아서 한 번에 전송!

// Netty 내부 동작
private void flush() {
    // outboundBuffer에 쌓인 여러 버퍼를 배열로 변환
    ByteBuf[] buffers = outboundBuffer.nioBuffers();
    // [header, body, footer]

    // 한 번의 시스템 콜로 전송 (Gathering Write)
    long bytesWritten = channel.write(buffers);
    // writev() 시스템 콜 1번으로 3개 버퍼 전송!
}

// HTTP 응답 전송
ctx.write(httpHeaders);     // 버퍼에 추가 (전송 X)
ctx.write(responseBody);    // 버퍼에 추가 (전송 X)
ctx.writeAndFlush(trailer); // 3개를 모아서 한 번에 전송!

// 내부적으로 writev() 시스템 콜 1번
// writev([httpHeaders, responseBody, trailer])

// Tomcat: write 호출마다 즉시 전송 시도
response.getOutputStream().write(header);  // write() 시스템 콜
response.getOutputStream().write(body);    // write() 시스템 콜
response.getOutputStream().write(footer);  // write() 시스템 콜
// → 시스템 콜 3번

// Netty: flush 전까지 버퍼에 모아둠
ctx.write(header);   // 버퍼에만 저장
ctx.write(body);     // 버퍼에만 저장
ctx.writeAndFlush(footer);  // writev() 1번으로 한 번에 전송
// → 시스템 콜 1번

클라이언트가 초당 10,000개 요청 전송 (빠름)
    ↓
Netty EventLoop가 요청을 읽어서 Handler로 전달
    ↓
Handler에서 DB 조회 (느림: 초당 100개만 처리 가능)
    ↓
읽은 데이터가 메모리에 계속 쌓임
    ↓
OutOfMemoryError 발생!

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    // 1. 요청 받음
    HttpRequest request = (HttpRequest) msg;

    // 2. DB 처리 중이면 읽기 중단
    if (dbQueue.size() > 100) {
        ctx.channel().config().setAutoRead(false);  // 네트워크 읽기 중단!
    }

    // 3. DB 조회 (비동기)
    CompletableFuture.supplyAsync(() ->
        dbRepository.find(request.id()), dbExecutor
    )
    .thenAccept(result -> {
        ctx.writeAndFlush(createResponse(result));

        // 4. DB 처리 완료 후 읽기 재개
        if (dbQueue.size() < 50) {
            ctx.channel().read();  // 네트워크 읽기 재개
        }
    });
}

[읽기 중단 전]
클라이언트 → Netty → Handler → DB 큐 (100개)
계속 읽음   요청 쌓임    느림      메모리 부족!

[읽기 중단 후]
클라이언트 → Netty (읽기 중단) → Handler → DB 큐 (50개로 감소)
대기        더 이상 안 읽음      처리 중         안정적

[읽기 재개]
클라이언트 → Netty (읽기 재개) → Handler → DB 큐 (여유 있음)
전송 재개       요청 읽음        정상 처리        안정적

Flux.range(1, 1000000)
    .onBackpressureBuffer(1000)  // 버퍼링: 1000개까지 버퍼에 저장
    .onBackpressureDrop()         // 드랍: 처리 못하면 버림
    .onBackpressureLatest()       // 최신 값만: 가장 최근 값만 유지
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            request(10);  // 10개만 요청
        }

        @Override
        protected void hookOnNext(Integer value) {
            // 처리
            request(1);  // 처리 완료 후 1개 더 요청
        }
    });

// EventLoopGroup - CPU 코어별 EventLoop
EventLoopGroup group = new NioEventLoopGroup(
    Runtime.getRuntime().availableProcessors() * 2
);

// 각 EventLoop는 하나의 Selector를 가짐
// → CPU 캐시 친화적

WebClient.subscribe()
    ↓ (메인 스레드)
Mono 체인 구성
    ↓
ReactorClientHttpConnector
    ↓
Netty HttpClient.request()
    ↓
Channel.write() → EventLoop에 제출
    ↓ (reactor-http-nio-2 스레드)
NioEventLoop.execute(task)
    ↓
selector.select() 호출
    ↓ (JNI)
epoll_wait() / kevent() 호출
    ↓ (OS 커널)
Ready List 모니터링
    ↓
네트워크 패킷 도착
    ↓
인터럽트 발생
    ↓
Ready List에 fd 추가
    ↓
epoll_wait() 반환
    ↓ (Java)
processSelectedKeys()
    ↓
channel.read() → ByteBuf
    ↓
Pipeline 처리 (HTTP 파싱)
    ↓
Mono.next(response)
    ↓
.doOnNext() 실행
    ↓
.subscribe() 콜백 실행