강의/Java

[실전자바] 멀티스레드와 동시성 (4)

atmosg 2025. 3. 25. 21:58

# 참고용 코드

  • 예시코드에 등장하는 자체 유틸리티 클래스들 적어둠
    더보기

    # ExecutorUtils - 스레드플 출력용

    public abstract class ExecutorUtils {
  
  public static void printState(ExecutorService es) {
    if (!(es instanceof ThreadPoolExecutor)) {
      LogUtils.info(es);
      return;
    }

    ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor) es;
    int poolSize = poolExecutor.getPoolSize();
    int activeCount = poolExecutor.getActiveCount();
    int queuedTasks = poolExecutor.getQueue().size();
    long completedTasks = poolExecutor.getCompletedTaskCount();

    StringBuilder sb = new StringBuilder();
    String str = sb.append("[thread in pool=").append(poolSize)
      .append(", activeCount=").append(activeCount)
      .append(", queuedTasks=").append(queuedTasks)
      .append(", completedTasks=").append(completedTasks)
      .toString();

    LogUtils.info(str);
  }

  public static void printState(ExecutorService es, String taskName) {
    if (!(es instanceof ThreadPoolExecutor)) {
      LogUtils.info(taskName + " -> " + es);
      return;
    }
    printState(es);
  }
  
}

     

     

    # LogUtils - 로그 출력용

    package utils;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;

public class LogUtils {

  public static void info(Object obj) {
    System.out.printf("[%s][%-9s] %s%n", 
      current(),
      Thread.currentThread().getName(),
      obj.toString()
    );
  }

  private static String current() {
    return LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss.SSS"));
  }

}

     


    # ThreadUtils - sleep 예외처리 귀찮아서 만든거

    package utils;

public abstract class ThreadUtils {
  
  public static void sleep(long milis) {
    try {
      Thread.sleep(milis);
    } catch (InterruptedException e) {
      LogUtils.info("인터럽트 발생 > " + e.getMessage());
      throw new RuntimeException(e);
    }
  }

}

     

 

 

 

12. Graceful Shutdown

# 우아한 종료란 (graceful shutdown)

  • 프로그램/서비스의 종료 시 수행 중이던 작업을 안전하고 깔끔하게 마무리한 후 종료하는 방식을 말함
  • 우아한 종료는 실무에서 매우 중요함. 기업에선 서버를 잠깐 종료하는 한 순간에도 돈과 관련된 일이 수없이 일어나기 때문
  • 그래서 종료를 어떻게 할 것인가에 대해 충분히 고민해봐야함
    • 이미 처리중이던 작업을 완료한 후 종료할 것인가 (근데 작업이 너무 많으면 어떡할래?)
    • 아니면 작업이고 나발이고 그냥 다 꺼버릴 것인가
    • 실행 대기열에 있던 작업은 어떻게 처리할 것인가(별도 보관? 그냥 폐기?)
    • 기타 등등 고려할 점이 많음


# ExecutorService 종료 메서드

  • ExecutorService에선 종료를 위한 메서드는 크게 세 가지가 지원됨
  • shutdown()

    • 신규 작업요청을 모두 쳐내고, 이미 제출된 작업들만 모두 완료 후 종료
    • 논 블로킹 메서드임 (즉, main스레드가 shoutdown 호출 했다치면, 호출 후에도 main스레드는 지 할일 계속함)
    • 자바19부턴 close()로 대체됐는데, 얘는 shutdown()하고 하루 기다려도 종료 안되면 shutdownNow()를 호출함

  • shutdownNow()

    • 말 그대로 강제 종룔르 위한 메서드
    • 실행 중인 작업은 인터럽트로 중지, 대기열에 대기중이던 작업은 List<Runnable> 형태로 반환함
    • 논 블로킹 메서드임

  • awaitTermination(long timeout, TimeUnit unit)

    • 모든 작업이 완료되길 기다린 후 종료하는 메서드
    • 단, 지정된 시간만큼만 대기
    • 블로킹 메서드임

  • 그림으로 이해하기
    더보기

    # shutdown()

     

     

    # shutdownNow()

     

     



 

# 우아한 종료 구현 - ExecutorService

  • 우아한 종료를 위해선 여러 상황을 고려해야함

    • 갑자기 요청이 너무 많이와서 큐에 대기중인 작업이 많음
    • 작업 자체가 그냥 너무 오래걸림
    • 버그가 발생해서 특정 작업이 끝나질 않음 등등
    • 결국, 가장 좋은 방법은 특정 시간까지 작업처리를 기다려보되 안되면 강제종료가 제일 현실적인 대안임

  • 아래는 ExecutorService 공식 문서에서 제안하는 방식임 (요약하자면 shutdown() → N초간 종료안되면 shutdownNow())
    더보기

    # 우아한 종료를 위한 메서드 구현

    package thread.executor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import utils.LogUtils;

public abstract class ExecutorUtils {
  
  public static void printState(ExecutorService es) {...}

  public static void printState(ExecutorService es, String taskName) {...}

  public static void shutdownAndAwaitTermination(ExecutorService es) {
    LogUtils.info("#shutdownAndAwaitTermination() > 서비스를 종료합니다.");
    // 논블로킹이라 아래코드도 계속 호출됨
    es.shutdown();
    
    try {
      LogUtils.info("서비스 종료 시도중... (최대 대기시간 8초)");
      
      // 얘는 블로킹
      boolean isShutdowned = es.awaitTermination(8, TimeUnit.SECONDS);
      if (!isShutdowned) {
        LogUtils.info("서비스 종료 실패 > 강제 종료 시도");
        
        // 얘도 논블로킹이라 아래코드도 계속 호출됨
        es.shutdownNow();
        boolean isRetrySuccess = es.awaitTermination(8, TimeUnit.SECONDS);
        if (!isRetrySuccess) {
          LogUtils.info("서비스 종료에 최종 실패하였습니다.");
        }
      }
    } catch (InterruptedException e) {
      // awaitTermination()으로 대기중인 현재 스레드가 인터럽트될 수 있음.
      es.shutdownNow();
    }

    LogUtils.info("#shutdownAndAwaitTermination() > 종료메서드 완료");
  }

}

     

    # 동작예시

    public class ShutdownMain {
  public static void main(String[] args) throws InterruptedException {
    ExecutorService es = Executors.newFixedThreadPool(2);
    es.execute(new RunnableTask("taskA", 1000));
    es.execute(new RunnableTask("taskB", 1000));
    es.execute(new RunnableTask("taskC", 1000));

    // 겁나 오래걸리는 작업업
    es.execute(new RunnableTask("taskD", 100_000));

    ExecutorUtils.printState(es);
    ExecutorUtils.shutdownAndAwaitTermination(es);
    ExecutorUtils.printState(es);
  }
  
}

     

     

    [21:21:21.148][main     ] [thread in pool=2, activeCount=2, queuedTasks=2, completedTasks=0
    [21:21:21.148][pool-1-thread-1] taskA 작업 시작
    [21:21:21.148][pool-1-thread-2] taskB 작업 시작
    [21:21:21.151][main     ] # shutdownAndAwaitTermination() > 서비스를 종료합니다.
    [21:21:21.152][main     ] 서비스 종료 시도중... (최대 대기시간 8초)
    [21:21:22.156][pool-1-thread-2] taskB 작업 종료
    [21:21:22.156][pool-1-thread-1] taskA 작업 종료
    [21:21:22.158][pool-1-thread-2] taskC 작업 시작
    [21:21:22.159][pool-1-thread-1] taskD 작업 시작
    [21:21:23.170][pool-1-thread-2] taskC 작업 종료
    [21:21:29.167][main     ] 서비스 종료 실패 > 강제 종료 시도
    [21:21:29.173][pool-1-thread-1] 인터럽트 발생 > sleep interrupted
    Exception in thread "pool-1-thread-1" 

    [21:21:29.174][main java.lang.RuntimeException: java.lang.InterruptedException: sleep interrupted]   at utils.ThreadUtils.sleep(ThreadUtils.java:10)
    # shutdownAndAwaitTermination() > 종료메서드 완료

     

 

 

 

13. Executor 스레드풀 관리

# 작업요청 거절 - RejectedExecutionException

  • 바로 이전 장에서 스레드풀도 꽉차고, 큐에 자리도 없을때 작업요청이 오면 거절한다고 설명한 바 있음
  • Executor의 스레드풀 관리는 다음과 같은 방식으로 동작함

    • 작업 요청이 들어오면 core 사이즈만큼 스레드 생성 및 작업할당
    • core사이즈를 초과하면 큐에 작업을 넣음
    • 큐 사이즈를 초과하면 maximum 사이즈만큼 비상용 스레드를 만듦
    • maximum 사이즈조차 초과해버리면 예외를 발생시켜서 작업요청을 거절함

  • 해당 과정이 실제로 어떻게 발생하는지 코드와 그림으로 확인해보자
    더보기

    # 그림으로 이해하기

    • corePoolSize=2, maximumPoolSize=3 (즉, 스레드는 비상동원 포함 최대 3개가 끝)
    • 대기열에 쓰일 블로킹큐의 최대 사이즈는 1

     

     

     

     

    # 동작예시

    package thread.graceful;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import thread.executor.ExecutorUtils;
import utils.LogUtils;
import utils.ThreadUtils;

public class Rejected {

  public static void main(String[] args) {
    BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(1);
    ExecutorService es = new ThreadPoolExecutor(2, 3, 3, TimeUnit.SECONDS, workQueue);

    ExecutorUtils.printState(es);

    LogUtils.info("##### 작업을 시작합니다 #####");
    for (int i=1; i<=5; i++) {
      String taskName = "task" + i;
      try {
        es.execute(new RunnableTask(taskName));
        ExecutorUtils.printState(es, taskName);
      } catch (RejectedExecutionException e) {
        LogUtils.info("# 실행거절 예외발생 > " + taskName + " > " + e);
      }
    }
    ThreadUtils.sleep(3000);
    LogUtils.info("#### 작업이 종료되었습니다 #####");
    
    ThreadUtils.sleep(3000);
    LogUtils.info("#### maximumPoolSize 대기시간 초과 #####");
    ExecutorUtils.printState(es);

    es.close();
    LogUtils.info("#### shutdown #####");
    ExecutorUtils.printState(es);
  }

}

     

     

    [21:54:02.958][main     ] [thread in pool=0, activeCount=0, queuedTasks=0, completedTasks=0
    [21:54:02.961][main     ] ##### 작업을 시작합니다 #####
    [21:54:02.965][main     ] [thread in pool=1, activeCount=1, queuedTasks=0, completedTasks=0
    [21:54:02.966][main     ] [thread in pool=2, activeCount=2, queuedTasks=0, completedTasks=0
    [21:54:02.966][pool-1-thread-1] task1 작업 시작
    [21:54:02.966][pool-1-thread-2] task2 작업 시작
    [21:54:02.966][main     ] [thread in pool=2, activeCount=2, queuedTasks=1, completedTasks=0
    [21:54:02.967][main     ] [thread in pool=3, activeCount=3, queuedTasks=1, completedTasks=0
    [21:54:02.967][pool-1-thread-3] task4 작업 시작
    [21:54:02.971][main     ] # 실행거절 예외발생 > task5 > java.util.concurrent.RejectedExecutionException: Task thread.graceful.RunnableTask@12edcd21 rejected from java.util.concurrent.ThreadPoolExecutor@a09ee92[Running, pool size = 3, active threads = 3, queued tasks = 1, completed tasks = 0]
    [21:54:03.973][pool-1-thread-3] task4 작업 종료
    [21:54:03.973][pool-1-thread-2] task2 작업 종료
    [21:54:03.973][pool-1-thread-1] task1 작업 종료
    [21:54:03.973][pool-1-thread-3] task3 작업 시작
    [21:54:04.976][pool-1-thread-3] task3 작업 종료
    [21:54:05.979][main     ] #### 작업이 종료되었습니다 #####
    [21:54:08.994][main     ] #### maximumPoolSize 대기시간 초과 #####
    [21:54:08.995][main     ] [thread in pool=2, activeCount=0, queuedTasks=0, completedTasks=4
    [21:54:08.996][main     ] #### shutdown #####
    [21:54:08.997][main     ] [thread in pool=0, activeCount=0, queuedTasks=0, completedTasks=4

     

 

 

 

# 스레드 풀 관리 전략

  • Executors 클래스에선 스레드 풀의 생성 및 관리를 위해 기본 세 가지 전략을 제공한다
  • 단일 스레드풀 전략 newSingleThreadPool()
    • 스레드 풀엔 기본 스레드 1개만을, 큐는 사이즈 제한이 없는 무제한 큐를 사용(LinkedBlockinQueue)
    • 주로 간단한 테스트 용도로 사용
  • 고정 스레드풀 전략 newFixedThreadPool(N)
    • N개의 기본 스레드를 생성하되, 비상동원 스레드는 생성하지 않는 방식
    • 큐 사이즈에 제한이 없음(LinkedBlockingQueue)
    • 스레드 수가 고정돼있어서 CPU, 메모리 사용량이 어느정도 예측 가능한 안정적인 방식임

  • 캐시 스레드풀 전략 newCachedThreadPool()
    • 기본 스레드는 사용X. 대신 60초 생존주기를 가진 초과 스레드만을 사용함
    • 작업 요청이 오는 족족 초과 스레드를 만들어서 처리하는 방식
    • 그래서 큐 사이즈도 0인 SynchronousQueue를 사용함
    • 모든 작업 요청이 큐에 대기하지 않고, 바로바로 스레드가 처리하므로 빠른 처리가 가능

  • 코드로 확인하기
    더보기 
    // 단일 스레드풀 전략
// 위 아래가 동일한 역할을 함
ExecutorService es = Executors.newSingleThreadPool();
ExecutorService es = new ThreadPoolExecutor(1, 1, 0L, 
 TimeUnit.MICROSECONDS, new LinkedBlockingQueue<>());

// 고정 스레드풀 전략
// 위 아래가 동일한 역할을 함
ExecutorService es = Executors.newFixedThreadPool(1000);
ExecutorService es = new ThreadPoolExecutor(1000, 1000, 0L, 
 TimeUnit.MICROSECONDS, new LinkedBlockingQueue<>());
 
// 캐시 스레드풀 전략
// 위 아래가 동일한 역할을 함
ExecutorService es = Executors.newFixedThreadPool(1000);
ExecutorService es = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, 
 TimeUnit.SECONDS, new SynchronousQueue<Runnable>());

 

SynchronousQueue

저장 공간이 0인 매우 특별한 큐. 소비자-생산자가 데이터를 직접 주고 받는 방식을 구현한다. 즉, 생산자가 데이터를 큐에 넣으려고 하면, 바로 가져갈 소비자가 나타날 때까지 대기함. 반대로 소비자가 데이터를 가져가려 하면, 데이터를 제공할 생산자가 나타날 때까지 대기함. (이를 즉시 전달 방식(Direct hand-off)라고 함)

 

 

 

# 관리 전략별 취약점과 권장방

  • 고정 스레드풀 전략 newFixedThreadPool(N)
    • 정해진 스레드만 쓰기 때문에 CPU, 메모리 사용량이 언제나 안정적이고 문제가 없어보임
    • 그런데 큐엔 100만건의 작업 처리 요청이 쌓여있을 수도 있음
    • 즉, 서비스가 커지거나 갑자기 사용자가 폭증하면 쌓인 요청대비 스레드가 적어서 응답속도가 늦어짐
    • 문제는 리소스 여유가 충분한데도 응답속도가 느려진다는 점임

  • 캐시 스레드풀 전략 newCachedThreadPool()
    • 스레드를 상황에 맞춰 무제한으로 생성해서 씀
    • 그러다보니 임계치 이상으로 사용자가 몰리면 CPU 100%, 메모리 100%로 서버가 터져버릴 수 있음
    • 실제로 스레드 1개당 약 1MB인걸 고려하면 사용자 10만명이 몰릴 시 메모리 100GB가 필요함

  • 위 두 취약점을 보완하고자 아래 방법을 권장함
    • new ThreadPoolExecutor(100, 200, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));
    • 기본스레드 100개, 비상동원 스레드 100개(60초 생존주기), 1000개 작업 대기가능한 큐
    • 이 방법이면 점진적인 서비스 확대 및 갑작스런 요청 증가에 어느정도 대응 가능함
    • 물론, 본인 서비스 상황에 맞춰 구체적인 스레드 수는 조정해야 겠지만
    • 예시 코드
      더보기

      # TASK_SIZE1, 2, 3별로 돌려가며 실행시켜보자

      package thread.graceful;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import thread.executor.ExecutorUtils;
import utils.LogUtils;

public class Startegy {
  
  static final int TASK_SIZE1 = 1100; // 스레드가 긴급동원되지 않음
  static final int TASK_SIZE2 = 1200; // 100개의 스레드가 긴급동원됨
  static final int TASK_SIZE3 = 1201; // 작업요청 1건에 대해 실행거부가 발생함

  public static void main(String[] args) {
    ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(1000);
    ExecutorService es = new ThreadPoolExecutor(100, 200, 60, TimeUnit.SECONDS, workQueue);

    ExecutorUtils.printState(es);
    long st = System.currentTimeMillis();
    
    for (int i=1; i<=TASK_SIZE1; i++) {
      String taskName = "task" + i;
      try {
        es.execute(new RunnableTask(taskName));
        ExecutorUtils.printState(es, taskName);
      } catch (RejectedExecutionException e) {
        LogUtils.info(taskName + " -> " + e);
      }
    }
    es.close();
    
    long ed = System.currentTimeMillis();
    ExecutorUtils.printState(es);
    LogUtils.info("time: " + (ed-st));
  }
}

 

 

# 스레드 미리 생성해두기

  • 요청에 의해 스레드를 생성하기 전에도 스레드풀에 미리 생성해둘 수 있음
  • 보통 응답시간이 아주 중요한 서버라면 이렇게 하는게 좋음
  • ThreadPoolExecutor 클래스의 prestartAllCoreThreads() 메서드가 제공함
  • 참고로 ExecutorService엔 이 메서드가 정의돼있지 않음
  • 코드로 확인하기
    더보기 
    public class PrestartPoolMain {
 public static void main(String[] args) {
   ExecutorService es = Executors.newFixedThreadPool(1000);
   ExecutorUtils.printState(es);
   
   ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor) es;
   poolExecutor.prestartAllCoreThreads();
   ExecutorUtils.printState(es);
 }
}

     

    [20:43:45.908][main     ] [thread in pool=0, activeCount=0, queuedTasks=0, completedTasks=0
    [20:43:45.942][main     ] [thread in pool=1000, activeCount=0, queuedTasks=0, completedTasks=0

 



14. Executor 예외정책

# 작업거절 방법

  • 실무에선 작업요청 과다로 서비스가 불가할 때 어떻게 예외처리할지 반드시 심사숙고해야함
    • 사용자에게 : 어떤 방식으로 서비스에 문제가 있음을 알릴 것인가
    • 개발자에게 : 어떻게 로그를 남기고 후속 조치할 것인가
    • 그냥 '오류발생: 관리자에게 문의하세요' 이딴 식으로 처리하면 진짜 박살남
  • Executor 프레임워크는 작업수용이 불가한 상황에서 사용할 수 있게끔 작업 거절에 대한 아래 정책들을 제공함
    • AbortPolicy : 기본 정책. 새 작업이 제출되면 RejectedExecutionException을 발생시킴
    • DiscardPolicy : 새 작업요청을 그냥 조용히 폐기하고 별도로 예외도 발생시키지 않
    • CallerRunsPolicy : 작업요청한 스레드가 직접 작업을 실행하도록 만들어버림
    • 사용자 정의 정책 : RejectedExecutionHandler를 구해 직접 거절 정책을 정의할 수도 있음. 사실 위 세 정책 모두 Executor 프레임워크가 사전에 미리  RejectedExecutionHandler를 구현해놓은 것들임


# 기본 정책 - AbortPolicy

  • 기본 정책으로, 더이상 작업수용이 불가한 상황에서 새 작업 제출시 RejectedExecutionException을 발생시킴
  • try ... catch 문을 활용해 catch 단계에 재시도든 작업포기든 알아서 하면 됨
    public class RejectMain {

  public static void main(String[] args) {
    SynchronousQueue<Runnable> workQueue = new SynchronousQueue<>();
    AbortPolicy abortPolicy = new ThreadPoolExecutor.AbortPolicy();
    ExecutorService es = new ThreadPoolExecutor(1,1,0, 
      TimeUnit.SECONDS, workQueue, abortPolicy);
    
    try {
      es.submit(new RunnableTask("task1", 1000));
      es.submit(new RunnableTask("task2", 1000));
    } catch (RejectedExecutionException e) {
      LogUtils.info("작업 요청 초과 > RejectedExecutionException 발생");
      LogUtils.info(e);
      // 이후 작업포기, 다시시도 등 상황에 맞게 작성
    }

    es.close();
  }

}

// [21:11:08.510][pool-1-thread-1] task1 작업 시작
// [21:11:08.510][main     ] 작업 요청 초과 > RejectedExecutionException 발생
// [21:11:08.514][main     ] RejectedExecutionException: Task FutureTask@681a9515[Not completed, task = RunnableAdapter@72ea2f77 ...]
// [21:11:09.521][pool-1-thread-1] task1 작업 종료

// 아래는 AboartPolicy 원본
public static class AbortPolicy implements RejectedExecutionHandler {
  
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +" rejected from " + e.toString());
  }
}

 

 

# 폐기 정책 - DiscardPolicy

  • 작업 요청을 무시하고 예외도 발생시키지 않는 정책
    public class RejectMain {

  public static void main(String[] args) {
    SynchronousQueue<Runnable> workQueue = new SynchronousQueue<>();
    DiscardPolicy discardPolicy = new ThreadPoolExecutor.DiscardPolicy();
    ExecutorService es = new ThreadPoolExecutor(1,1,0, 
      TimeUnit.SECONDS, workQueue, discardPolicy);
    
    try {
      es.submit(new RunnableTask("task1", 1000));
      es.submit(new RunnableTask("task2", 1000));
    } catch (RejectedExecutionException e) {
      LogUtils.info("작업 요청 초과 > RejectedExecutionException 발생");
      LogUtils.info(e);
      // 이후 작업포기, 다시시도 등 상황에 맞게 작성
    }

    es.close();
  }

}


// [21:39:46.120][pool-1-thread-1] task1 작업 시작
// [21:39:47.133][pool-1-thread-1] task1 작업 종료
// 아래는 DiscardPolicy 원본
public static class DiscardPolicy implements RejectedExecutionHandler {
  
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    // empty
  }
  
}

 

 

# 폐기 정책 - DiscardPolicy

  • 작업을 요청한 주체가 직접 처리하도록 만들어버리는 정책
  • '작업요청좀 작작해라' 라는 의미로 작업을 직접 처리하게해서 작업요청 속도를 느리게 만들어버림
  • 별도로 예외를 발생시키지 않음
    public class RejectMain {

  public static void main(String[] args) {
    SynchronousQueue<Runnable> workQueue = new SynchronousQueue<>();
    CallerRunsPolicy callerRunsPolicy = new ThreadPoolExecutor.CallerRunsPolicy();
    ExecutorService es = new ThreadPoolExecutor(1,1,0, 
      TimeUnit.SECONDS, workQueue, callerRunsPolicy);
    
    try {
      es.submit(new RunnableTask("task1", 1000));
      es.submit(new RunnableTask("task2", 1000));
      es.submit(new RunnableTask("task3", 1000));
      es.submit(new RunnableTask("task4", 1000));
    } catch (RejectedExecutionException e) {
      LogUtils.info("작업 요청 초과 > RejectedExecutionException 발생");
      LogUtils.info(e);
      // 이후 작업포기, 다시시도 등 상황에 맞게 작성
    }

    es.close();
  }

}


[21:45:42.826][pool-1-thread-1] task1 작업 시작
[21:45:42.826][main     ] task2 작업 시작 <-- 작업을 요청한 main 스레드가 직접 처리함
[21:45:43.833][main     ] task2 작업 종료
[21:45:43.833][pool-1-thread-1] task1 작업 종료
[21:45:43.834][main     ] task3 작업 시작
[21:45:44.835][main     ] task3 작업 종료
[21:45:44.836][pool-1-thread-1] task4 작업 시작
[21:45:45.851][pool-1-thread-1] task4 작업 종료
// 아래는 CallerRunsPolicy 원본
public static class CallerRunsPolicy implements RejectedExecutionHandler {
  
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) r.run();
  }
  
}

 

 

# 사용자 정의 정책

  • RejectedExecutionHandler 인터페이스를 구현해서 입맛대로 만들면 됨
  • 아래는 거절된 작업을 버리되, 경고 로그를 남겨서 개발자가 인지할 수 있게끔 만든 예시코드임
    더보기 
    public class RejectMain {

  public static void main(String[] args) {
    SynchronousQueue<Runnable> workQueue = new SynchronousQueue<>();
    ExecutorService es = new ThreadPoolExecutor(1,1,0, 
      TimeUnit.SECONDS, workQueue, new CustomRejectPolicy());
    
    try {
      es.submit(new RunnableTask("task1", 1000));
      es.submit(new RunnableTask("task2", 1000));
      es.submit(new RunnableTask("task3", 1000));
      es.submit(new RunnableTask("task4", 1000));
    } catch (RejectedExecutionException e) {
      LogUtils.info("작업 요청 초과 > RejectedExecutionException 발생");
      LogUtils.info(e);
      // 이후 작업포기, 다시시도 등 상황에 맞게 작성
    }

    es.close();
  }

  static class CustomRejectPolicy implements RejectedExecutionHandler {

    static AtomicInteger count = new AtomicInteger(0);

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
      int i = count.incrementAndGet();
      LogUtils.info("[Warn] 거절된 누적 작업 수 : " + i);
    }

  }

}

     

    [21:57:53.093][pool-1-thread-1] task1 작업 시작
    [21:57:53.093][main     ] [Warn] 거절된 누적 작업 수 : 1
    [21:57:53.095][main     ] [Warn] 거절된 누적 작업 수 : 2
    [21:57:53.096][main     ] [Warn] 거절된 누적 작업 수 : 3
    [21:57:54.100][pool-1-thread-1] task1 작업 종료