private final static int CONCURRENCY_CALL = 10;
List<RefineResult> results = Flux.fromIterable(registerDtos)
.flatMap(dto -> Mono.defer(() ->
omsService.refineAddress(dto.getPrimaryAddress(), dto.getSecondaryAddress())
.defaultIfEmpty(ApiResponse.failure(“NO_RESPONSE”, false))
).map(resp -> new RefineResult(dto, resp)), CONCURRENCY_CALL)
.collectList()
.block();

@Override
public Mono<ApiResponse<RefineAddressDto>> refineAddress(String primaryAddress, String secondaryAddress) {
    RefineAddressInput request = RefineAddressInput.builder()
        .primaryAddress(primaryAddress)
        .secondaryAddress(secondaryAddress)
        .build();
<span class="k">return</span> <span class="n">omsClient</span><span class="o">.</span><span class="na">post</span><span class="o">()</span>
    <span class="o">.</span><span class="na">uri</span><span class="o">(</span><span class="s">"/refine-address"</span><span class="o">)</span>
    <span class="o">.</span><span class="na">bodyValue</span><span class="o">(</span><span class="n">request</span><span class="o">)</span>
    <span class="o">.</span><span class="na">retrieve</span><span class="o">()</span>
    <span class="o">.</span><span class="na">bodyToMono</span><span class="o">(</span><span class="nc">RefineAddressOutput</span><span class="o">.</span><span class="na">class</span><span class="o">)</span>
    <span class="o">.</span><span class="na">timeout</span><span class="o">(</span><span class="nc">Duration</span><span class="o">.</span><span class="na">ofSeconds</span><span class="o">(</span><span class="mi">5</span><span class="o">))</span>
    <span class="o">.</span><span class="na">retryWhen</span><span class="o">(</span><span class="nc">RetryPolicy</span><span class="o">.</span><span class="na">fixedDelay</span><span class="o">(</span><span class="mi">1</span><span class="o">,</span> <span class="nc">Duration</span><span class="o">.</span><span class="na">ofMillis</span><span class="o">(</span><span class="mi">100</span><span class="o">),</span> <span class="s">"[OMS 주소정제] 재시도 요청: "</span> <span class="o">+</span> <span class="n">request</span><span class="o">))</span>
    <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="n">output</span> <span class="o">-&gt;</span> <span class="n">output</span><span class="o">.</span><span class="na">isSuccess</span><span class="o">()</span> <span class="o">?</span> <span class="nc">ApiResponse</span><span class="o">.</span><span class="na">success</span><span class="o">(</span><span class="nc">RefineAddressDto</span><span class="o">.</span><span class="na">from</span><span class="o">(</span><span class="n">output</span><span class="o">))</span>
        <span class="o">:</span> <span class="nc">ApiResponse</span><span class="o">.&lt;</span><span class="nc">RefineAddressDto</span><span class="o">&gt;</span><span class="n">failure</span><span class="o">(</span><span class="s">"응답 결과에 데이터가 없음"</span><span class="o">,</span> <span class="kc">false</span><span class="o">))</span>
    <span class="o">.</span><span class="na">onErrorResume</span><span class="o">(</span><span class="n">ex</span> <span class="o">-&gt;</span> <span class="nc">ExternalErrorHandler</span><span class="o">.</span><span class="na">handleError</span><span class="o">(</span><span class="n">ex</span><span class="o">,</span> <span class="n">extractOmsErrorMessage</span><span class="o">(</span><span class="n">ex</span><span class="o">),</span> <span class="s">"OMS 주소정제"</span><span class="o">));</span>

}

List<RefineResult> results = Flux.fromIterable(registerDtos)
    .flatMap(dto -> Mono.defer(() -> ...))
    .collectList()
    .block();  // ← 구독 시작점

block() → collectList() → flatMap() → Mono.defer() → WebClient 체인

// 구독될 때마다 새로운 WebClient 체인 생성
Mono.defer(() -> omsService.refineAddress(...))

omsClient.post()
    .uri("/refine-address")
    .bodyValue(request)
    .retrieve()
    .bodyToMono(RefineAddressOutput.class)

// 개념적인 코드
while (true) {
    // 1. 네트워크에서 뭔가 일어났는지 확인
    네트워크_이벤트_확인();
    // 2. 일어난 일들 처리
    이벤트들_처리();
    // 3. 누가 시켜놓은 작업들 처리
    작업큐에서_작업꺼내서_실행();
}

// SingleThreadIoEventLoop.java:153-164
protected void run() {
    do {
        runIo();                    // ← 1+2: I/O 확인 및 처리
        runAllTasks(maxTasksPerRun); // ← 3: 작업큐 처리
    } while (!confirmShutdown());
}

// NioIoHandler.java:420-485
public int run(IoExecutionContext runner) {
    // 1단계: select - I/O 이벤트 존재 여부 확인
    select(runner, wakenUp.getAndSet(false));
<span class="c1">// 2단계: 있으면 처리</span>
<span class="k">return</span> <span class="nf">processSelectedKeys</span><span class="o">();</span>

}

// NioIoHandler.java
private void select(IoExecutionContext runner, boolean oldWakenUp) {
    Selector selector = this.selector;
<span class="k">for</span> <span class="o">(;;)</span> <span class="o">{</span>
    <span class="c1">// 태스크가 있으면 즉시 확인하고 넘어감</span>
    <span class="k">if</span> <span class="o">(!</span><span class="n">runner</span><span class="o">.</span><span class="na">canBlock</span><span class="o">()</span> <span class="o">&amp;&amp;</span> <span class="n">wakenUp</span><span class="o">.</span><span class="na">compareAndSet</span><span class="o">(</span><span class="kc">false</span><span class="o">,</span> <span class="kc">true</span><span class="o">))</span> <span class="o">{</span>
        <span class="n">selector</span><span class="o">.</span><span class="na">selectNow</span><span class="o">();</span>   <span class="c1">// 작업 있으면 바로 확인</span>
        <span class="k">break</span><span class="o">;</span>
    <span class="o">}</span>

    <span class="c1">// 태스크가 없으면 이벤트 올 때까지 대기</span>
    <span class="kt">int</span> <span class="n">selectedKeys</span> <span class="o">=</span> <span class="n">selector</span><span class="o">.</span><span class="na">select</span><span class="o">(</span><span class="n">timeoutMillis</span><span class="o">);</span>
<span class="o">}</span>

}

@Override
public boolean canBlock() {
   assert inEventLoop();
   return !hasTasks() && !hasScheduledTasks();
}

// SelectorImpl.java
@Override
    public final int select(long timeout) throws IOException {
        return lockAndDoSelect(null, (timeout == 0) ? -1 : timeout);
    }

// SelectorImpl.java
private int lockAndDoSelect(Consumer<SelectionKey> action, long timeout)
      throws IOException
{
    synchronized (this) {
        ensureOpen();
        if (inSelect)
            throw new IllegalStateException("select in progress");
        inSelect = true;
        try {
            synchronized (publicSelectedKeys) {
                return doSelect(action, timeout);
            }
        } finally {
            inSelect = false;
        }
    }
}

EPollSelectorImpl(SelectorProvider sp) throws IOException {
   super(sp);
// 1. epoll 인스턴스 생성 (epoll_create 시스템 콜)
this.epfd = EPoll.create();
// 2. epoll_wait 결과를 저장할 네이티브 메모리 할당
this.pollArrayAddress = EPoll.allocatePollArray(NUM_EPOLLEVENTS);
// 3. wakeup용 EventFD 생성
this.eventfd = new EventFD();
IOUtil.configureBlocking(IOUtil.newFD(eventfd.efd()), false);
// 4. EventFD를 epoll에 EPOLLIN으로 등록
EPoll.ctl(epfd, EPOLL_CTL_ADD, eventfd.efd(), EPOLLIN);
}

// EpollSelectorImpl.java
@Override
protected int doSelect(Consumer<SelectionKey> action, long timeout) throws IOException {
    int to = (int) Math.min(timeout, Integer.MAX_VALUE);
<span class="kt">int</span> <span class="n">numEntries</span><span class="o">;</span>
<span class="n">processUpdateQueue</span><span class="o">();</span>      <span class="c1">// epoll_ctl로 관심 이벤트 변경 반영</span>
<span class="n">processDeregisterQueue</span><span class="o">();</span>

<span class="k">try</span> <span class="o">{</span>
    <span class="n">begin</span><span class="o">(</span><span class="n">blocking</span><span class="o">);</span>
    <span class="c1">// epoll_wait 시스템 콜 호출</span>
    <span class="n">numEntries</span> <span class="o">=</span> <span class="nc">EPoll</span><span class="o">.</span><span class="na">wait</span><span class="o">(</span><span class="n">epfd</span><span class="o">,</span> <span class="n">pollArrayAddress</span><span class="o">,</span> <span class="no">NUM_EPOLLEVENTS</span><span class="o">,</span> <span class="n">to</span><span class="o">);</span>
<span class="o">}</span> <span class="k">finally</span> <span class="o">{</span>
    <span class="n">end</span><span class="o">(</span><span class="n">blocking</span><span class="o">);</span>
<span class="o">}</span>

<span class="c1">// 반환된 이벤트 처리</span>
<span class="k">return</span> <span class="nf">processEvents</span><span class="o">(</span><span class="n">numEntries</span><span class="o">,</span> <span class="n">action</span><span class="o">);</span>

}

 // NioIoHandler.java
  private int processSelectedKeysOptimized() {
      int handled = 0;
      for (int i = 0; i < selectedKeys.size; ++i) {
          SelectionKey k = selectedKeys.keys[i];
          selectedKeys.keys[i] = null;  // GC를 위해 null 처리
      <span class="n">processSelectedKey</span><span class="o">(</span><span class="n">k</span><span class="o">);</span>  <span class="c1">// 각 이벤트 처리</span>
      <span class="o">++</span><span class="n">handled</span><span class="o">;</span>
  <span class="o">}</span>
  <span class="k">return</span> <span class="n">handled</span><span class="o">;</span>

}

  private void processSelectedKey(SelectionKey k) {
      final DefaultNioRegistration registration = (DefaultNioRegistration) k.attachment();
  <span class="c1">// 준비된 이벤트를 핸들러에 전달</span>
  <span class="c1">// OP_READ  → 데이터 수신</span>
  <span class="c1">// OP_WRITE → 데이터 송신</span>
  <span class="c1">// OP_CONNECT → 연결 완료</span>
  <span class="c1">// OP_ACCEPT → 새 연결 요청</span>
  <span class="n">registration</span><span class="o">.</span><span class="na">handle</span><span class="o">(</span><span class="n">k</span><span class="o">.</span><span class="na">readyOps</span><span class="o">());</span>

}

// AbstractNioChannel.java:420-450
@Override
public void handle(IoRegistration registration, IoEvent event) {
    NioIoOps nioReadyOps = ((NioIoEvent) event).ops();
<span class="c1">// 1. OP_CONNECT: 연결 완료 처리 (가장 먼저 처리)</span>
<span class="k">if</span> <span class="o">(</span><span class="n">nioReadyOps</span><span class="o">.</span><span class="na">contains</span><span class="o">(</span><span class="nc">NioIoOps</span><span class="o">.</span><span class="na">CONNECT</span><span class="o">))</span> <span class="o">{</span>
    <span class="n">removeAndSubmit</span><span class="o">(</span><span class="nc">NioIoOps</span><span class="o">.</span><span class="na">CONNECT</span><span class="o">);</span>
    <span class="n">unsafe</span><span class="o">().</span><span class="na">finishConnect</span><span class="o">();</span>
<span class="o">}</span>

<span class="c1">// 2. OP_WRITE: 쓰기 가능 상태 - 대기 중인 버퍼 전송</span>
<span class="k">if</span> <span class="o">(</span><span class="n">nioReadyOps</span><span class="o">.</span><span class="na">contains</span><span class="o">(</span><span class="nc">NioIoOps</span><span class="o">.</span><span class="na">WRITE</span><span class="o">))</span> <span class="o">{</span>
    <span class="n">forceFlush</span><span class="o">();</span>
<span class="o">}</span>

<span class="c1">// 3. OP_READ / OP_ACCEPT: 데이터 수신 또는 새 연결 수락</span>
<span class="k">if</span> <span class="o">(</span><span class="n">nioReadyOps</span><span class="o">.</span><span class="na">contains</span><span class="o">(</span><span class="nc">NioIoOps</span><span class="o">.</span><span class="na">READ_AND_ACCEPT</span><span class="o">)</span> <span class="o">||</span> <span class="n">nioReadyOps</span><span class="o">.</span><span class="na">equals</span><span class="o">(</span><span class="nc">NioIoOps</span><span class="o">.</span><span class="na">NONE</span><span class="o">))</span> <span class="o">{</span>
    <span class="n">read</span><span class="o">();</span>
<span class="o">}</span>

}

// SingleThreadEventExecutor.java
protected boolean runAllTasks(long timeoutNanos) {
     // 스케줄 큐에서 실행 가능한 태스크를 TaskQueue로 이동
     fetchFromScheduledTaskQueue();
     Runnable task = pollTask();
 <span class="kd">final</span> <span class="kt">long</span> <span class="n">deadline</span> <span class="o">=</span> <span class="n">timeoutNanos</span> <span class="o">&gt;</span> <span class="mi">0</span> <span class="o">?</span> <span class="n">getCurrentTimeNanos</span><span class="o">()</span> <span class="o">+</span> <span class="n">timeoutNanos</span> <span class="o">:</span> <span class="mi">0</span><span class="o">;</span>
 <span class="kt">long</span> <span class="n">runTasks</span> <span class="o">=</span> <span class="mi">0</span><span class="o">;</span>

 <span class="k">for</span> <span class="o">(;;)</span> <span class="o">{</span>
     <span class="n">safeExecute</span><span class="o">(</span><span class="n">task</span><span class="o">);</span> <span class="c1">// 테스크 실행</span>

     <span class="n">runTasks</span> <span class="o">++;</span>

        <span class="c1">// Check timeout every 64 tasks because nanoTime() is relatively expensive.</span>
        <span class="c1">// XXX: Hard-coded value - will make it configurable if it is really a problem.</span>
        <span class="k">if</span> <span class="o">((</span><span class="n">runTasks</span> <span class="o">&amp;</span> <span class="mh">0x3F</span><span class="o">)</span> <span class="o">==</span> <span class="mi">0</span><span class="o">)</span> <span class="o">{</span>
            <span class="n">lastExecutionTime</span> <span class="o">=</span> <span class="n">getCurrentTimeNanos</span><span class="o">();</span>
            <span class="k">if</span> <span class="o">(</span><span class="n">lastExecutionTime</span> <span class="o">&gt;=</span> <span class="n">deadline</span><span class="o">)</span> <span class="o">{</span>
                <span class="k">break</span><span class="o">;</span>
            <span class="o">}</span>
        <span class="o">}</span>

        <span class="n">task</span> <span class="o">=</span> <span class="n">pollTask</span><span class="o">();</span>
        <span class="k">if</span> <span class="o">(</span><span class="n">task</span> <span class="o">==</span> <span class="kc">null</span><span class="o">)</span> <span class="o">{</span>
            <span class="n">lastExecutionTime</span> <span class="o">=</span> <span class="n">getCurrentTimeNanos</span><span class="o">();</span>
            <span class="k">break</span><span class="o">;</span>
        <span class="o">}</span>
    <span class="o">}</span>

    <span class="n">afterRunningAllTasks</span><span class="o">();</span>
    <span class="k">return</span> <span class="kc">true</span><span class="o">;</span>
<span class="o">}</span>

return omsClient.post()
   .uri("/refine-address")
   .bodyValue(request)
   .retrieve()
   .bodyToMono(RefineAddressOutput.class)
   .timeout(Duration.ofSeconds(5)) // ← 여기
   .retryWhen(