멀티 스레드와 동시성 (6) 스레드풀과 Executor 프레임워크

// 복잡한 결과 수집 방식 class SumTask implements Runnable { private int result; // 결과 저장용 필드 private final int start, end; @Override public void run() { result = calculateSum(start, end); } public int getResult() { return result; // 별도 메서드로 결과 조회 } } // 사용 시 SumTask task = new SumTask(1, 100); Thread thread = new Thread(task); thread.start(); thread.join(); // 완료 대기 int result = task.getResult(); // 결과 수집

public void run() { try { Thread.sleep(1000); // 체크 예외를 내부에서 처리해야 함 } catch (InterruptedException e) { // 강제로 내부 처리 } // throws InterruptedException 불가능 }

public interface Callable<V> { V call() throws Exception; // 반환값과 예외 선언 가능 }

// 간결한 Callable 구현 Callable<Integer> sumTask = () -> { Thread.sleep(1000); // 체크 예외 자동 처리 return IntStream.rangeClosed(1, 100).sum(); // 직접 반환 }; // ExecutorService와 함께 사용 ExecutorService executor = Executors.newSingleThreadExecutor(); Future<Integer> future = executor.submit(sumTask); Integer result = future.get(); // 결과 직접 수집

ExecutorService executor = Executors.newSingleThreadExecutor(); // 1. 작업 제출 (논블로킹) Future<String> future = executor.submit(() -> { Thread.sleep(3000); // 3초 소요 작업 return "작업 완료!"; }); // 2. 즉시 반환되어 다른 작업 수행 가능 System.out.println("다른 작업 수행 중..."); doOtherWork(); // 3. 필요할 때 결과 조회 (블로킹) String result = future.get(); // 3초 후 결과 반환

// 여러 작업을 동시에 제출 List<Future<Integer>> futures = new ArrayList<>(); for (int i = 0; i < 10; i++) { final int taskId = i; Future<Integer> future = executor.submit(() -> heavyCalculation(taskId)); futures.add(future); // 즉시 반환되어 다음 작업 제출 가능 } // 모든 결과 수집 List<Integer> results = new ArrayList<>(); for (Future<Integer> future : futures) { results.add(future.get()); // 각 작업 완료 시점에 결과 수집 }

Future<String> future = executor.submit(() -> { if (someCondition) { throw new IllegalStateException("작업 실패"); } return "성공"; }); try { String result = future.get(); } catch (ExecutionException e) { Throwable cause = e.getCause(); // 원본 예외 추출 if (cause instanceof IllegalStateException) { // 구체적인 예외 처리 } }

try { // 최대 5초 대기 String result = future.get(5, TimeUnit.SECONDS); System.out.println("결과: " + result); } catch (TimeoutException e) { System.out.println("시간 초과로 작업을 취소합니다."); future.cancel(true); // 실행 중인 스레드에 인터럽트 } catch (ExecutionException e) { System.out.println("실행 오류: " + e.getCause().getMessage()); }

Future<String> future = executor.submit(() -> { for (int i = 0; i < 1000; i++) { if (Thread.currentThread().isInterrupted()) { throw new InterruptedException("작업이 취소되었습니다."); } // 무거운 작업 수행 heavyWork(i); } return "완료"; }); // 3초 후 취소 Thread.sleep(3000); boolean cancelled = future.cancel(true); // 인터럽트로 강제 취소

[요청 스레드] → [작업 큐] → [스레드 풀] → [작업 처리] (생산자). (BlokingQueue) (소비자)

public ThreadPoolExecutor( int corePoolSize, // 풀에서 관리되는 기본 스레드 수 int maximumPoolSize, // 풀에서 관리될 수 있는 최대 스레드 수 long keepAliveTime, // corePoolSize를 초과하여 생성된 스레드(초과 스레드) 생존 시간 TimeUnit unit, // 시간 단위 BlockingQueue<Runnable> workQueue, // 작업 큐 RejectedExecutionHandler handler // 거절 정책 )

public void gracefulShutdown(ExecutorService executor) { try { // 1단계: 우아한 종료 시도 executor.shutdown(); // 2단계: 지정 시간 동안 완료 대기 if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { // 3단계: 강제 종료 List<Runnable> pendingTasks = executor.shutdownNow(); System.out.println("강제 종료됨. 미완료 작업 수: " + pendingTasks.size()); // 4단계: 강제 종료 후 재확인 if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { System.err.println("스레드풀이 완전히 종료되지 않았습니다."); } } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } }

try (ExecutorService es = Executors.newFixedThreadPool(8)) { // 작업 제출 } // 여기서 close()가 호출되며 내부적으로 종료를 기다림

ExecutorService executor = Executors.newFixedThreadPool(10); // corePoolSize = maximumPoolSize = 10 // workQueue = LinkedBlockingQueue (무제한)

ExecutorService executor = Executors.newCachedThreadPool(); // corePoolSize = 0 // maximumPoolSize = Integer.MAX_VALUE (무제한) // workQueue = SynchronousQueue (저장 공간 없음)

ThreadPoolExecutor executor = new ThreadPoolExecutor( 5, // corePoolSize: 평상시 5개 스레드 20, // maximumPoolSize: 최대 20개 스레드 60L, // keepAliveTime: 초과 스레드 60초 생존 TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), // 큐 크기 100 제한 new ThreadPoolExecutor.CallerRunsPolicy() // 거절 시 호출자가 직접 실행 );

// 기본 정책으로 별도 설정 없이 사용 ThreadPoolExecutor executor = new ThreadPoolExecutor(...); // RejectedExecutionException 발생

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());

executor.setRejectedExecutionHandler(new RejectedExecutionHandler() { private final AtomicLong rejectedCount = new AtomicLong(0); @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { long count = rejectedCount.incrementAndGet(); // 로깅 logger.warn("작업 거절됨. 총 거절 횟수: {}", count); // 메트릭 수집 meterRegistry.counter("thread.pool.rejected").increment(); // 알림 또는 대체 처리 로직 if (count % 100 == 0) { alertService.sendAlert("스레드풀 거절 횟수가 " + count + "회에 도달했습니다."); } } })