Java의 ExecutorService로 스레드 풀 관리 🚀

Java 개발자라면 멀티스레딩의 중요성을 잘 알고 계실 겁니다. 특히 대규모 애플리케이션을 다룰 때, 효율적인 스레드 관리는 성능 최적화의 핵심이 됩니다. 이런 맥락에서 ExecutorService는 Java의 동시성 프로그래밍에서 매우 중요한 역할을 합니다. 🔧

이 글에서는 ExecutorService를 활용한 스레드 풀 관리에 대해 심도 있게 다루겠습니다. 기본 개념부터 고급 기술까지, 실제 개발 현장에서 바로 적용할 수 있는 실용적인 내용으로 구성했습니다. 재능넷과 같은 플랫폼에서 활동하는 개발자분들께 특히 유용할 것입니다.

자, 그럼 본격적으로 ExecutorService의 세계로 들어가 볼까요? 🏊‍♂️

1. ExecutorService 개요 📚

ExecutorService는 Java의 java.util.concurrent 패키지에 포함된 인터페이스로, 비동기적인 태스크 실행을 위한 프레임워크를 제공합니다. 이는 Thread 클래스를 직접 다루는 것보다 더 높은 수준의 추상화를 제공하여, 개발자가 복잡한 스레드 관리 로직에서 벗어나 비즈니스 로직에 집중할 수 있게 해줍니다.

1.1 ExecutorService의 주요 특징

• 스레드 풀 관리: 재사용 가능한 스레드 풀을 생성하고 관리합니다.
• 작업 큐: 실행할 작업들을 큐에 저장하고 관리합니다.
• 비동기 실행: submit() 메소드를 통해 작업을 비동기적으로 실행할 수 있습니다.
• Future 객체: 비동기 작업의 결과를 Future 객체를 통해 받을 수 있습니다.
• 다양한 종료 옵션: shutdown(), shutdownNow() 등의 메소드로 ExecutorService를 안전하게 종료할 수 있습니다.

1.2 ExecutorService vs Thread

전통적인 Thread 클래스와 비교했을 때, ExecutorService는 다음과 같은 이점을 제공합니다:

이러한 특징들로 인해 ExecutorService는 현대적인 Java 애플리케이션에서 널리 사용되고 있습니다. 특히 웹 서버, 데이터베이스 연결 풀, 그리고 재능넷과 같은 대규모 온라인 플랫폼에서 중요한 역할을 합니다.

다음 섹션에서는 ExecutorService의 기본 사용법에 대해 자세히 알아보겠습니다. 🔍

2. ExecutorService 기본 사용법 🛠️

ExecutorService를 사용하기 위해서는 먼저 인스턴스를 생성해야 합니다. Java는 다양한 종류의 ExecutorService 구현체를 제공하며, 각각 다른 특성을 가지고 있습니다.

2.1 ExecutorService 생성하기

가장 일반적인 ExecutorService 생성 방법은 Executors 클래스의 팩토리 메소드를 사용하는 것입니다:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// 고정 크기 스레드 풀
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);

// 캐시 스레드 풀
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

// 단일 스레드 실행자
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();

// 스케줄링 가능한 스레드 풀
ExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  

2.2 작업 제출하기

ExecutorService에 작업을 제출하는 방법은 크게 두 가지가 있습니다:

2.2.1 execute() 메소드

execute() 메소드는 Runnable 인터페이스를 구현한 작업을 실행합니다. 이 메소드는 작업의 결과를 반환하지 않습니다.

executorService.execute(() -> {
    System.out.println("작업 실행 중...");
    // 작업 로직
});
  

2.2.2 submit() 메소드

submit() 메소드는 Callable 또는 Runnable 인터페이스를 구현한 작업을 실행하고, Future 객체를 반환합니다. Future를 통해 작업의 결과를 얻거나 작업의 상태를 확인할 수 있습니다.

Future<String> future = executorService.submit(() -> {
    // 작업 로직
    return "작업 완료";
});

// 결과 얻기
try {
    String result = future.get();
    System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}
  

2.3 ExecutorService 종료하기

ExecutorService를 더 이상 사용하지 않을 때는 반드시 종료해야 합니다. 그렇지 않으면 프로그램이 종료되지 않을 수 있습니다.

executorService.shutdown();  // 새로운 작업 수락을 중지하고 기존 작업을 마무리

// 또는

executorService.shutdownNow();  // 즉시 종료를 시도하고 미처리 작업 목록을 반환
  

안전한 종료를 위해 다음과 같은 패턴을 사용할 수 있습니다:

executorService.shutdown();
try {
    if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
        executorService.shutdownNow();
    }
} catch (InterruptedException e) {
    executorService.shutdownNow();
}
  

이 패턴은 먼저 shutdown()을 호출하여 새로운 작업의 수락을 중지하고, 60초 동안 기존 작업의 완료를 기다립니다. 60초 후에도 작업이 완료되지 않으면 shutdownNow()를 호출하여 강제 종료를 시도합니다.

이렇게 ExecutorService의 기본적인 사용법을 살펴보았습니다. 다음 섹션에서는 더 심화된 내용으로, 다양한 ExecutorService 구현체와 그 특징에 대해 알아보겠습니다. 💡

3. ExecutorService 구현체와 특징 🔍

Java는 다양한 ExecutorService 구현체를 제공합니다. 각 구현체는 서로 다른 특징을 가지고 있어, 애플리케이션의 요구사항에 따라 적절한 구현체를 선택할 수 있습니다.

3.1 FixedThreadPool

FixedThreadPool은 고정된 수의 스레드를 가진 스레드 풀입니다.

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
  

• 특징:
  • 지정된 수의 스레드를 생성하고 유지합니다.
  • 스레드가 사용 가능해질 때까지 작업을 큐에 저장합니다.
  • 장기 실행 작업에 적합합니다.
• 사용 사례: CPU 바운드 작업, 일정한 처리량이 필요한 경우

3.2 CachedThreadPool

CachedThreadPool은 필요에 따라 스레드를 생성하고 재사용하는 유연한 스레드 풀입니다.

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
  

• 특징:
  • 필요에 따라 새 스레드를 생성합니다.
  • 이전에 생성된 스레드를 재사용합니다.
  • 60초 동안 사용되지 않은 스레드는 제거됩니다.
• 사용 사례: 많은 단기 비동기 작업, I/O 바운드 작업

3.3 SingleThreadExecutor

SingleThreadExecutor는 단일 스레드로 작업을 순차적으로 실행합니다.

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
  

• 특징:
  • 단일 스레드로 모든 작업을 순차적으로 실행합니다.
  • 작업의 순서를 보장합니다.
• 사용 사례: 순차적 실행이 필요한 작업, 공유 리소스 접근

3.4 ScheduledThreadPool

ScheduledThreadPool은 일정 시간 후 또는 주기적으로 작업을 실행할 수 있는 스레드 풀입니다.

ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  

• 특징:
  • 지정된 시간 후에 작업을 실행할 수 있습니다.
  • 주기적인 작업 실행이 가능합니다.
• 사용 사례: 주기적인 백그라운드 작업, 타이머 작업

3.5 WorkStealingPool (Java 8+)

WorkStealingPool은 ForkJoinPool을 기반으로 한 작업 훔치기(work-stealing) 알고리즘을 사용하는 스레드 풀입니다.

ExecutorService workStealingPool = Executors.newWorkStealingPool();
  

• 특징:
  • 사용 가능한 프로세서 수에 따라 병렬 수준을 자동으로 조정합니다.
  • 작업 훔치기 알고리즘으로 효율적인 작업 분배를 수행합니다.
• 사용 사례: 병렬 처리가 필요한 대규모 작업, 재귀적 작업

각 ExecutorService 구현체는 고유한 특징과 장단점을 가지고 있습니다. 애플리케이션의 요구사항, 작업의 특성, 그리고 시스템 리소스를 고려하여 적절한 구현체를 선택해야 합니다. 예를 들어, 재능넷과 같은 플랫폼에서 다양한 사용자 요청을 처리할 때는 CachedThreadPool이 유용할 수 있습니다. 반면, 정기적인 데이터 백업과 같은 작업에는 ScheduledThreadPool이 적합할 것입니다.

다음 섹션에서는 ExecutorService를 사용할 때 주의해야 할 점과 최적화 전략에 대해 알아보겠습니다. 🚀

4. ExecutorService 사용 시 주의사항 및 최적화 전략 ⚠️

ExecutorService는 강력한 도구이지만, 잘못 사용하면 성능 저하나 예기치 않은 문제를 일으킬 수 있습니다. 여기서는 ExecutorService를 효과적으로 사용하기 위한 주의사항과 최적화 전략을 살펴보겠습니다.

4.1 스레드 풀 크기 설정

스레드 풀의 크기는 성능에 큰 영향을 미칩니다. 너무 작으면 병렬 처리의 이점을 살리지 못하고, 너무 크면 리소스 낭비와 컨텍스트 스위칭 오버헤드가 발생합니다.

• CPU 바운드 작업: 일반적으로 사용 가능한 코어 수 + 1로 설정
• I/O 바운드 작업: 코어 수보다 많이 설정 (예: 코어 수 * 2)

int coreCount = Runtime.getRuntime().availableProcessors();
ExecutorService executorService = Executors.newFixedThreadPool(coreCount + 1);
  

4.2 작업 제출 및 실행

작업을 제출할 때는 다음 사항을 고려해야 합니다:

• 작업의 크기와 복잡성을 고려하여 적절한 단위로 나누기
• long-running 작업은 별도의 스레드 풀로 분리
• 작업 간 의존성 관리 (CompletableFuture 활용)

4.3 예외 처리

ExecutorService에서 발생하는 예외를 적절히 처리하지 않으면 스레드가 종료되어 풀의 크기가 줄어들 수 있습니다.

executorService.submit(() -> {
    try {
        // 작업 로직
    } catch (Exception e) {
        logger.error("작업 실행 중 오류 발생", e);
    }
});
  

4.4 리소스 관리

ExecutorService를 사용할 때는 리소스 관리에 주의해야 합니다:

• 사용이 끝난 ExecutorService는 반드시 종료 (shutdown)
• 메모리 누수 방지를 위해 작업 내에서 사용한 리소스 해제
• ThreadLocal 변수 사용 시 주의 (스레드 재사용으로 인한 데이터 오염 가능성)

4.5 모니터링 및 튜닝

성능 최적화를 위해 ExecutorService의 동작을 모니터링하고 튜닝해야 합니다:

• JMX를 통한 스레드 풀 상태 모니터링
• 로깅을 통한 작업 실행 시간 및 처리량 추적
• 프로파일링 도구를 사용한 병목 지점 식별

4.6 최적화 전략

ExecutorService 사용을 최적화하기 위한 몇 가지 전략을 소개합니다:

1. 작업 분할 (Task Splitting): 큰 작업을 작은 단위로 나누어 병렬 처리 효율을 높입니다.
2. 작업 우선순위 지정: PriorityBlockingQueue를 사용하여 중요한 작업을 먼저 처리합니다.
3. 백프레셔 (Backpressure) 구현: 작업 제출 속도를 조절하여 시스템 과부하를 방지합니다.
4. 캐시 활용: 자주 사용되는 계산 결과를 캐시하여 중복 작업을 줄입니다.
5. 비동기 프로그래밍 패턴 활용: CompletableFuture를 사용하여 복잡한 비동기 워크플로우를 구현합니다.

이러한 전략들을 적절히 조합하여 사용하면 ExecutorService의 성능을 크게 향상시킬 수 있습니다. 예를 들어, 재능넷과 같은 플랫폼에서 사용자 요청을 처리할 때, 작업 분할과 우선순위 지정을 통해 중요한 요청을 빠르게 처리하면서도 전체적인 처리량을 높일 수 있습니다.

다음 섹션에서는 ExecutorService를 사용한 실제 구현 예제를 통해 이러한 개념들을 어떻게 적용할 수 있는지 살펴보겠습니다. 🛠️

5. ExecutorService 실제 구현 예제 💻

이 섹션에서는 ExecutorService를 사용한 실제 구현 예제를 통해 지금까지 배운 개념들을 어떻게 적용할 수 있는지 살펴보겠습니다. 재능넷과 같은 온라인 플랫폼에서 사용할 수 있는 시나리오를 바탕으로 예제를 구성했습니다.

5.1 사용자 프로필 업데이트 시스템

재능넷에서 여러 사용자의 프로필을 동시에 업데이트해야 하는 상황을 가정해 보겠습니다. 이 작업은 데이터베이스 접근과 외부 API 호출 등이 포함될 수 있는 I/O 바운드 작업입니다.

import java.util.List;
import java.util.concurrent.*;

public class UserProfileUpdater {
    private final ExecutorService executorService;

    public UserProfileUpdater(int threadPoolSize) {
        this.executorService = Executors.newFixedThreadPool(threadPoolSize);
    }

    public void updateProfiles(List<User> users) {
        List<CompletableFuture<Void>> futures = users.stream()
            .map(user -> CompletableFuture.runAsync(() -> updateProfile(user), executorService))
            .collect(Collectors.toList());

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }

    private void updateProfile(User user) {
        try {
            // 프로필 업데이트 로직
            System.out.println("Updating profile for user: " + user.getId());
            // 데이터베이스 업데이트
            // 외부 API 호출
            Thread.sleep(1000); // 시뮬레이션을 위한 지연
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            System.err.println("Error updating profile for user: " + user.getId());
            e.printStackTrace();
        }
    }

    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
        }
    }

    // User 클래스 (간단한 예시)
    static class User {
        private final int id;

        User(int id) {
            this.id = id;
        }

        public int getId() {
            return id;
        }
    }

    public static void main(String[] args) {
        int threadPoolSize = Runtime.getRuntime().availableProcessors() * 2; // I/O 바운드 작업이므로 코어 수의 2배로 설정
        UserProfileUpdater updater = new UserProfileUpdater(threadPoolSize);

        List<User> users = IntStream.range(0, 100)
            .mapToObj(User::new)
            .collect(Collectors.toList());

        long startTime = System.currentTimeMillis();
        updater.updateProfiles(users);
        long endTime = System.currentTimeMillis();

        System.out.println("Total time taken: " + (endTime - startTime) + "ms");

        updater.shutdown();
    }
}
  

이 예제에서는 다음과 같은 ExecutorService 최적화 전략을 적용했습니다:

• I/O 바운드 작업에 적합한 스레드 풀 크기 설정 (코어 수의 2배)
• CompletableFuture를 사용한 비동기 작업 처리
• 예외 처리를 통한 안정성 확보
• 안전한 ExecutorService 종료

5.2 병렬 데이터 처리 시스템

다음은 재능넷에서 대량의 사용자 데이터를 병렬로 처리하는 시나리오입니다. 이 예제에서는 작업 분할 전략을 사용하여 대규모 데이터를 효율적으로 처리합니다.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ParallelDataProcessor {
    private final ExecutorService executorService;
    private final int batchSize;

    public ParallelDataProcessor(int threadPoolSize, int batchSize) {
        this.executorService = Executors.newFixedThreadPool(threadPoolSize);
        this.batchSize = batchSize;
    }

    public List<ProcessedData> processData(List<RawData> rawDataList) {
        List<List<RawData>> batches = splitIntoBatches(rawDataList);

        List<Future<List<ProcessedData>>> futures = batches.stream()
            .map(batch -> executorService.submit(() -> processBatch(batch)))
            .collect(Collectors.toList());

        List<ProcessedData> result = new ArrayList<>();
        for (Future<List<ProcessedData>> future : futures) {
            try {
                result.addAll(future.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

        return result;
    }

    private List<List<RawData>> splitIntoBatches(List<RawData> rawDataList) {
        return IntStream.range(0, rawDataList.size())
            .boxed()
            .collect(Collectors.groupingBy(index -> index / batchSize))
            .values()
            .stream()
            .map(indices -> indices.stream()
                .map(rawDataList::get)
                .collect(Collectors.toList()))
            .collect(Collectors.toList());
    }

    private List<ProcessedData> processBatch(List<RawData> batch) {
        return batch.stream()
            .map(this::processData)
            .collect(Collectors.toList());
    }

    private ProcessedData processData(RawData rawData) {
        // 데이터 처리 로직
        try {
            Thread.sleep(100); // 처리 시간 시뮬레이션
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return new ProcessedData(rawData.getId(), "Processed: " + rawData.getData());
    }

    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
        }
    }

    // 데이터 클래스들
    static class RawData {
        private final int id;
        private final String data;

        RawData(int id, String data) {
            this.id = id;
            this.data = data;
        }

        public int getId() { return id; }
        public String getData() { return data; }
    }

    static class ProcessedData {
        private final int id;
        private final String processedData;

        ProcessedData(int id, String processedData) {
            this.id = id;
            this.processedData = processedData;
        }

        @Override
        public String toString() {
            return "ProcessedData{id=" + id + ", data='" + processedData + "'}";
        }
    }

    public static void main(String[] args) {
        int threadPoolSize = Runtime.getRuntime().availableProcessors();
        int batchSize = 100;
        ParallelDataProcessor processor = new ParallelDataProcessor(threadPoolSize, batchSize);

        List<RawData> rawDataList = IntStream.range(0, 10000)
            .mapToObj(i -> new RawData(i, "Data " + i))
            .collect(Collectors.toList());

        long startTime = System.currentTimeMillis();
        List<ProcessedData> processedDataList = processor.processData(rawDataList);
        long endTime = System.currentTimeMillis();

        System.out.println("Processed " + processedDataList.size() + " items");
        System.out.println("Total time taken: " + (endTime - startTime) + "ms");

        processor.shutdown();
    }
}
  

이 예제에서 적용된 주요 전략들은 다음과 같습니다:

• 작업 분할: 대량의 데이터를 작은 배치로 나누어 처리
• 병렬 처리: ExecutorService를 사용하여 배치를 동시에 처리
• 결과 수집: Future를 사용하여 비동기 작업의 결과를 수집
• 리소스 관리: 작업 완료 후 ExecutorService를 안전하게 종료

이러한 예제들은 ExecutorService를 실제 상황에서 어떻게 활용할 수 있는지 보여줍니다. 재능넷과 같은 플랫폼에서는 이러한 기술을 사용하여 대량의 사용자 데이터를 효율적으로 처리하고, 시스템의 전반적인 성능을 향상시킬 수 있습니다.

다음 섹션에서는 ExecutorService 사용 시 자주 발생하는 문제점들과 그 해결 방법에 대해 알아보겠습니다. 🔧

6. ExecutorService 사용 시 주의점 및 문제 해결 🚨

ExecutorService를 사용하면서 개발자들이 자주 겪는 문제점들과 그 해결 방법에 대해 알아보겠습니다.

6.1 스레드 풀 고갈 (Thread Pool Exhaustion)

문제: 모든 스레드가 장기 실행 작업으로 인해 블로킹되어 새로운 작업을 처리할 수 없는 상황

해결방법:

• 적절한 스레드 풀 크기 설정
• 타임아웃 메커니즘 구현
• 장기 실행 작업을 별도의 스레드 풀로 분리

ExecutorService longRunningTaskPool = Executors.newFixedThreadPool(5);
ExecutorService quickTaskPool = Executors.newFixedThreadPool(20);

// 장기 실행 작업
longRunningTaskPool.submit(() -> {
    // 장시간 실행되는 작업
});

// 빠른 작업
quickTaskPool.submit(() -> {
    // 짧은 시간 내에 완료되는 작업
});
  

6.2 메모리 누수 (Memory Leaks)

문제: 작업 내에서 사용된 리소스가 제대로 해제되지 않아 발생하는 메모리 누수

해결방법:

• try-with-resources 구문 사용
• 작업 완료 후 명시적으로 리소스 해제
• WeakReference 사용 고려

executorService.submit(() -> {
    try (FileInputStream fis = new FileInputStream("file.txt")) {
        // 파일 처리 로직
    } catch (IOException e) {
        e.printStackTrace();
    }
});
  

6.3 예외 처리 미흡

문제: 작업 내에서 발생한 예외가 적절히 처리되지 않아 스레드가 종료되는 상황

해결방법:

• UncaughtExceptionHandler 구현
• 작업 내에서 try-catch 블록 사용
• Future.get() 메소드 사용 시 예외 처리

Thread.setDefaultUncaughtExceptionHandler((t, e) -> {
    System.err.println("Uncaught exception in thread " + t.getName() + ": " + e.getMessage());
});

Future<?> future = executorService.submit(() -> {
    // 작업 로직
});

try {
    future.get();
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}
  

6.4 데드락 (Deadlock)

문제: 여러 스레드가 서로의 작업 완료를 기다리며 무한정 블로킹되는 상황

해결방법:

• 락 획득 순서 일관성 유지
• 타임아웃 설정
• 순환 의존성 제거

ReentrantLock lock1 = new ReentrantLock();
ReentrantLock lock2 = new ReentrantLock();

executorService.submit(() -> {
    try {
        if (lock1.tryLock(1, TimeUnit.SECONDS)) {
            try {
                if (lock2.tryLock(1, TimeUnit.SECONDS)) {
                    try {
                        // 두 락을 모두 획득한 후의 로직
                    } finally {
                        lock2.unlock();
                    }
                }
            } finally {
                lock1.unlock();
            }
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});
  

6.5 과도한 컨텍스트 스위칭

문제: 너무 많은 스레드 생성으로 인한 과도한 컨텍스트 스위칭 발생

해결방법:

• 적절한 스레드 풀 크기 설정
• 작업 크기 조정 (너무 작은 작업 지양)
• Work Stealing 알고리즘 사용 (ForkJoinPool)

int parallelism = Runtime.getRuntime().availableProcessors();
ExecutorService workStealingPool = Executors.newWorkStealingPool(parallelism);

// 큰 작업을 재귀적으로 분할하여 처리
workStealingPool.submit(() -> processLargeTask(data));
  

이러한 문제점들을 인식하고 적절한 해결 방법을 적용함으로써, ExecutorService를 더욱 안정적이고 효율적으로 사용할 수 있습니다. 재능넷과 같은 대규모 플랫폼에서는 이러한 문제들이 시스템 전체의 성능과 안정성에 큰 영향을 미칠 수 있으므로, 개발 초기 단계부터 이러한 문제들을 고려하여 설계하는 것이 중요합니다.

다음 섹션에서는 ExecutorService의 고급 사용법과 최신 Java 버전에서 제공하는 관련 기능들에 대해 알아보겠습니다. 🚀

7. ExecutorService의 고급 사용법 및 최신 동향 🔬

ExecutorService의 더 깊은 이해와 효과적인 활용을 위해, 고급 사용법과 Java의 최신 버전에서 제공하는 관련 기능들을 살펴보겠습니다.

7.1 CompletableFuture와의 통합

Java 8에서 도입된 CompletableFuture는 비동기 프로그래밍을 더욱 강력하게 만들어줍니다. ExecutorService와 함께 사용하면 복잡한 비동기 워크플로우를 쉽게 구현할 수 있습니다.

ExecutorService executorService = Executors.newFixedThreadPool(4);

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 비동기 작업
    return "결과";
}, executorService).thenApplyAsync(result -> {
    // 결과 처리
    return result + " 처리됨";
}, executorService);

future.thenAccept(System.out::println);
  

7.2 Fork/Join 프레임워크

Fork/Join 프레임워크는 대규모 작업을 작은 작업으로 분할하고 병렬로 처리한 후 결과를 합치는 "분할 정복" 알고리즘을 구현하는 데 특화되어 있습니다.

ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());

class SumTask extends RecursiveTask<Long> {
    private final long[] numbers;
    private final int start;
    private final int end;

    // 생성자 및 기타 메소드 생략

    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= 1000) {
            return sum();
        }

        SumTask leftTask = new SumTask(numbers, start, start + length / 2);
        leftTask.fork();

        SumTask rightTask = new SumTask(numbers, start + length / 2, end);
        Long rightResult = rightTask.compute();
        Long leftResult = leftTask.join();

        return leftResult + rightResult;
    }
}

long result = forkJoinPool.invoke(new SumTask(numbers, 0, numbers.length));
  

7.3 Virtual Threads (Java 19+)

Java 19에서 프리뷰 기능으로 도입된 Virtual Threads는 경량 스레드로, 기존의 플랫폼 스레드보다 훨씬 적은 리소스를 사용합니다. 이를 통해 수많은 동시 작업을 효율적으로 처리할 수 있습니다.

// Java 19+
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    IntStream.range(0, 10_000).forEach(i -> {
        executor.submit(() -> {
            Thread.sleep(Duration.ofSeconds(1));
            return i;
        });
    });
}
  

7.4 Reactive Streams

Reactive Streams API는 비동기 스트림 처리를 위한 표준을 제공합니다. ExecutorService와 함께 사용하여 백프레셔를 관리하고 비동기 데이터 흐름을 효과적으로 제어할 수 있습니다.

ExecutorService executorService = Executors.newFixedThreadPool(4);

Publisher<Integer> publisher = subscriber -> {
    subscriber.onSubscribe(new Subscription() {
        @Override
        public void request(long n) {
            executorService.submit(() -> {
                for (int i = 0; i < n; i++) {
                    subscriber.onNext(i);
                }
            });
        }

        @Override
        public void cancel() {
            // 취소 로직
        }
    });
};

publisher.subscribe(new Subscriber<Integer>() {
    // Subscriber 구현
});
  

7.5 Project Loom

현재 개발 중인 Project Loom은 Java의 동시성 모델을 크게 개선할 것으로 예상됩니다. 이 프로젝트는 경량 스레드(Fibers)와 구조적 동시성을 도입하여, 높은 동시성을 요구하는 애플리케이션의 개발을 더욱 쉽게 만들 것입니다.

// Project Loom (예상 코드, 실제 구현은 다를 수 있음)
try (var scope = new StructuredTaskScope()) {
    Future<String> future1 = scope.fork(() -> fetchDataFromApi1());
    Future<String> future2 = scope.fork(() -> fetchDataFromApi2());

    scope.join();  // 모든 작업이 완료될 때까지 대기

    String result1 = future1.resultNow();
    String result2 = future2.resultNow();

    System.out.println(result1 + " " + result2);
}
  

이러한 고급 기능들과 최신 동향을 이해하고 적절히 활용함으로써, 재능넷과 같은 대규모 플랫폼에서 더욱 효율적이고 확장 가능한 시스템을 구축할 수 있습니다. 특히 Virtual Threads와 Project Loom은 향후 Java의 동시성 프로그래밍에 큰 변화를 가져올 것으로 예상되므로, 이에 대한 지속적인 관심과 학습이 필요합니다.

마지막으로, ExecutorService의 미래와 Java 동시성 프로그래밍의 발전 방향에 대해 간략히 살펴보겠습니다. 🔮