CompletableFuture와 ForkJoinPool

비동기 프로그래밍의 두 축

Java 8 부터 도입된 CompletableFuture와 ForkJoinPool은 비동기 및 병렬 프로그래밍에서 중요한 역할을 합니다. 이 두 가지는 서로 보완적인 관계에 있으며, 효율적인 멀티스레드 프로그래밍을 가능하게 합니다. 

 

이번 글에서는 이들이 어떻게 함께 상호 보완적으로 동작하는지와 주의해야 될 점에 대해 살펴보겠습니다. 

 

CompletableFuture

CompletableFuture는 기존의 Future 인터페이스가 제공하지 못했던 비동기 결과 값의 조합과 예외 처리를 훨씬 더 효과적으로 수행할 수 있는 기능을 갖춘 인터페이스입니다. 이 인터페이스를 활용하면 동기적인 수행 방식을 비동기적인 수행 방식으로 변환하여 성능을 개선하고, 더 유연한 비동기 처리 로직을 구현할 수 있습니다.

 

예제

아래 예제는 ItemRepository에서 storeName을 통해 item을 조회하는 예제입니다. 

@Getter
public class Item {
	
	private String name;
	private Integer price;
	
	@Builder 
	public Item(String name, Integer price) {
		this.name = name;
		this.price = price;
	}
	
}

 

ItemRepository.getPriceByName(...) 메서드는 요청시마다 1초간 대기 후 가게에 등록되어 있는 Item의 가격을 리턴합니다. 

@Component
public class ItemRepository {
	
	private Map<String, Item> storeMap = new ConcurrentHashMap<>();
	
	public ItemRepository() {
		storeMap.put("storeA", Item.builder().name("pencil").price(300).build());
		storeMap.put("storeB", Item.builder().name("eraser").price(500).build());
		storeMap.put("storeC", Item.builder().name("mechanical pencil").price(2000).build());
	}
	
	public int getPriceByStoreName(String name) {
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		return storeMap.get(name).getPrice();
	}
	
}

 

해당 로직에서 동기적으로 모든 상품을 조회하면 시간이 얼마나 걸릴까요? 

@Test
public void getPriceAllSync() {
	int priceA = itemRepository.findPriceByStoreName("storeA");
	int priceB = itemRepository.findPriceByStoreName("storeB");
	int priceC = itemRepository.findPriceByStoreName("storeC");
	
	log.info("price = {}", priceA);
	log.info("price = {}", priceB);
	log.info("price = {}", priceC);
}

3sec 482ms 경과

findPriceByStoreName(...) 메서드는 호출될 때 마다 1초 동안 대기한 후 작업을 수행하기 때문에 작업 시간이 3초 이상 걸리게 됩니다. 이러한 방식의 로직은 여러 개의 스레드로 한 번에 요청을 한다면 경과 시간을 크게 줄일 수 있습니다. 

 

그렇다면 해당 로직을 비동기로 처리하도록 작성하려면 어떻게 해야할까요? 

@Test
public void getPriceAllAsync() {
	CompletableFuture<Void> futureA = CompletableFuture.supplyAsync(() -> {
		log.info("ThreadName = {}", Thread.currentThread().getName());
		return itemRepository.getPriceStoreName("storeA");
	  })
	  .thenAccept(price -> log.info("price = {}", price));
	  
	CompletableFuture<Void> futureB = CompletableFuture.supplyAsync(() -> {
		log.info("ThreadName = {}", Thread.currentThread().getName());
		return itemRepository.getPriceStoreName("storeB");
	  })
	  .thenAccept(price -> log.info("price = {}", price));
	  
	CompletableFuture<Void> futureC = CompletableFuture.supplyAsync(() -> {
		log.info("ThreadName = {}", Thread.currentThread().getName());
		return itemRepository.getPriceStoreName("storeC");
	  })
	  .thenAccept(price -> log.info("price = {}", price));
	
	CompletableFuture.allOf(futureA, futureB, futureC).join();
}

현재 로직에서는 개별 작업을 CompletableFuture.supplyAsync(...)를 사용하여 비동기적으로 처리하고 있습니다. 이 방식은 현재 스레드와는 별개의 스레드에서 함수가 실행되도록 하여 병렬로 작업을 수행할 수 있습니다. 

• theAccept(...): 결과를 소비하는 콜백함수로, 리턴타입이 없습니다. CompletableFuture.supplyAsync(...)에서의 작업이 완료된 후 결과를 처리하는 작업을 지정할 때 사용됩니다. 만약 리턴 타입이 있는 콜백 함수를 사용하고 싶다면, thenApply(...) 메서드를 사용해야 합니다. 
• allOf(...): 여러 CompletableFuture를 병렬로 처리할 수 있도록 해주는 메서드입니다. 모든 CompletableFuture의 처리가 완료될 때까지 대기하며, 완료된 후 CompletableFuture<Void>를 반환합니다. 그러나 allOf 메서드는 모든 Future의 결과를 결합한 값을 반환하지 않습니다. 

따라서 개별 CompletableFuture의 결과를 사용하려면 join() 메서드를 활용해야 합니다. join()을 호출하여 allOf()의 작업이 완료될 때까지 블로킹하여 대기할 수 있습니다. 

 

실행 결과 분석

3sec 482ms -> 1sec 450ms

일단 비동기로 실행한 코드가 이전 동기적으로 실행한 코드보다 빠른 수행 시간을 보인다는 것을 확인할 수 있습니다. 

 

위의 실행 결과를 보면, 해당 CompletableFuture가 실행된 스레드가 ForkJoinPool.commonPool로 표시되어 있습니다. 여기서 ForkJoinPool.commonPool은 무엇이고 왜 비동기 로직이 해당 스레드에서 동작하는 것일까요? 

 

ForkJoinPool

ForkJoinPool은 자바에서 제공하는 비동기 프레임워크로, 분할-정복 알고리즘을 활용하여 작업을 효율적으로 분산하고 병렬로 실행하는 스레드 풀입니다. 이 스레드 풀은 큰 작업을 여러 작은 Sub Task로 분리(Fork)하고, 각각을 병렬로 처리한 후 그 결과를 다시 합쳐(Join) 최종결과를 생성하는 방식으로 동작합니다. 

 

동작 방식

ForkJoinPool은 다른 종류의 ExecutorService와는 다른 Work-Stealing 매커니즘을 사용합니다.

ForkJoinPool의 동작방식

1. submit()을 통해 task를 보낸다.
2. 인바운드 큐에 task가 들어가고, A와 B 스레드가 task를 처리한다.
3. A와 B는 각자 큐가 있으며, 위의 그림의 B처럼 큐에 task가 없으면 A의 task를 steal하여 처리한다. 

Work-Stealing 매커니즘을 사용하기 때문에 CPU 자원이 놀지 않고 최적의 성능을 낼 수 있게 해 줍니다. 

 

ForkJoinPool과 CompletableFuture의 연계

Java 8에서부터 CompletableFuture는 ForkJoinPool을 사용하여 비동기 작업을 처리합니다. 기본적으로 CompletableFuture는 ForkJoinPool.commonPool()을 사용하여 스레드 풀을 가져오고 사용합니다. 

ForkJoinPool.commonPool() 디폴트 스레드 개수 = 코어 개수 * 2 - 1

ForkJoinPool.commonPool()은 기본적으로 사용 가능한 프로세서의 수에 따라 동적으로 크기가 조정되는 공용 ForkJoinPool을 반환합니다. 따라서 CompletableFuture를 사용할 때는 별도의 스레드 풀을 생성하거나 구성할 필요가 없으며, 기본적으로 시스템 리소스를 잘 활용할 수 있습니다. 

 

ForkJoinPool 사용 시 주의사항

CompletableFuture와 parallelStream은 내부적으로 ForkJoinPool.commonPool()을 사용하여 작업을 병렬화 시킵니다. 이 말은 commonPool을 별도의 스레드 풀로 생성하는 것이 아니라는 말과 같습니다. 즉, 별도의 설정이 없다면 하나의 스레드 풀을 모든 CompletableFuture과 parallelStream이 공유하게 되고, 스레드 풀을 사용하는 다른 스레드에 영향을 줄 수 있으며, 반대로 영향을 받을 수도 있습니다. 

 

아래와 같은 코드를 동시에 요청하여 실행한다고 가정해봅시다. 

public void getPriceAllAsync() {
	CompletableFuture<Integer> futureA = CompletableFuture.supplyAsync(() -> {
		log.info("Thread Name = {}", Thread.currentThread().getName());
		return itemRepository.getPriceByStoreName("storeA");
	});
	
	CompletableFuture<Integer> futureB = CompletableFuture.supplyAsync(() -> {
		log.info("Thread Name = {}", Thread.currentThread().getName());
		return itemRepository.getPriceByStoreName("storeB");
	});
	
	CompletableFuture<Integer> futureC = CompletableFuture.supplyAsync(() -> {
		log.info("Thread Name = {}", Thread.currentThread().getName());
		return itemRepository.getPriceByStoreName("storeC");
	});
	
	log.info("price = {}", futureA.join());
	log.info("price = {}", futureB.join());
	log.info("price = {}", futureC.join());
}

 첫 번째 요청이 해당 코드를 실행하였을 때 스레드 풀은 아래와 같은 상태가 됩니다. 

첫 번째 요청 실행

첫 번째 요청을 처리하는 동안 스레드를 점유하게 되면, 다음 요청이 도착할 때 스레드를 반환하지 않고 계속 점유하게 될 수 있습니다. 이로 인해 Thread 1, 2, 3은 사용할 수 없게 되고, 결국 Thread 4 하나만 남아 모든 요청을 처리해야 할 수 있습니다. 

 

특히 스레드가 sleep과 같이 아무 작업도 하지 않고 점유 중이면 심각한 문제가 발생할 수 있습니다. 

두 번째 요청 실행

위 과정에서 더 심각한 상황은 Thread 4까지 점유한 경우입니다. 이 경우 추가 요청은 처리되지 않고 Thread Pool Queue에 쌓이게 되며, 일정 시간이 지나면 요청이 손실될 수 있습니다. 

 

ForkJoinPool.commPool()은 전역적으로 공유되는 스레드 풀을 사용합니다. 이 때문에 4개의 스레드가 모두 사용 중이라면, 다른 작업들은 대기 상태에 들어가게 됩니다. 특히 Blocking I/O가 발생하는 작업이 있을 경우, 스레드 풀 내부의 스레드들이 블록 되어 다른 요청들이 스레드를 얻기 전까지 기다려야 합니다. 이로 인해 성능 저하와 같은 문제가 발생할 수 있습니다. 

 

public void getPriceAllAsync() {
	ExecutorService pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
	
	CompletableFuture<Integer> futureA = CompletableFuture.supplyAsync(() -> {
		log.info("Thread Name = {}", Thread.currentThread().getName());
		return itemRepository.getPriceByStoreName("storeA");
	}, pool);
	
	CompletableFuture<Integer> futureB = CompletableFuture.supplyAsync(() -> {
		log.info("Thread Name = {}", Thread.currentThread().getName());
		return itemRepository.getPriceByStoreName("storeB");
	}, pool);
	
	CompletableFuture<Integer> futurec = CompletableFuture.supplyAsync(() -> {
		log.info("Thread Name = {}", Thread.currentThread().getName());
		return itemRepository.getPriceByStoreName("storeC");
	}, pool);
	
	log.info("price = {}", futureA.join());
	log.info("price = {}", futureB.join());
	log.info("price = {}", futureC.join());
	
	pool.shutdown();
}

이 문제를 해결하기 위해, 각 CompletableFuture나 parallelStream에 대해 독립적인 스레드 풀을 사용하는 방법이 있습니다. 이렇게 하면 스레드 풀이 독립적으로 관리되므로, 한 작업이 스레드를 차지하고 있어도 다른 작업은 영향을 받지 않고 원활하게 실행할 수 있습니다. 

 

CompletableFuture와 parallelStream

Java에서 병렬처리를 하는 방법은 CompletableFuture만 있는 것이 아닌 parallelStream도 존재합니다. 이 둘은 ForkJoinPool.commonPool()을 공유하고 병렬 처리를 지원하지만, 각각의 특성과 적합한 사용 상황이 다릅니다. 

 

parallelStream

parallelStream은 Java 8부터 추가된 스트림 API의 확장으로 스트림을 병렬로 처리할 수 있도록 합니다. 이를 통해 데이터 집합을 병렬로 처리하여 성능을 향상 시킬 수 있습니다. 

 

아래는 이전 예제를 parallelStream을 통해 처리하는 예제입니다. 

@Test
public void getPriceAllAsyncWithParallelStream() {
	List<String> stores = List.of("storeA", "storeB", "storeC");
	
	stores.parallelStream() 
		.forEach(store -> {
			log.info("Thread Name = {}", Thread.currentThread().getName());
			log.info("price = {}", itemRepository.getPriceByStore(store));
		});
}

위의 코드를 실행해보면 이전 CompletableFuture를 사용한 방식과 다른 점 없이 동일한 결과와 수행 시간이 나타납니다. 그렇다면 CompletableFuture와 parallelStream 사이에는 어떤 차이가 있는 걸까요? 

 

parallelStream도 Stream이다

아래 예제를 통해 이 두 방식의 차이를 살펴보겠습니다. 

@Test 
public void getPriceAllAsyncWithParallelStream() throws Exception {
	List<String> stores = List.of("storeA", "storeB", "storeC");
	
	List<Integer> prices = stores.parallelStream()
		.map(store -> itemRepository.getPriceByStoreName(store))
		.toList();
	
	Thread.sleep(1000);
	
	prices.forEach(price -> log.info("price = {}", price));
}

이 코드는 parallelStream을 통한 가격 조회 부분과 해당 가격을 출력하는 코드 사이에 어떠한 작업이 있다고 가정하고 작성한 코드입니다. 

 

해당 코드를 수행한다면 아래와 같은 결과를 얻을 수 있습니다. 

수행시간 2sec 465ms

 

CompletableFuture를 사용한 코드도 같은 상황을 가정하여 작성하였습니다. 

@Test
public void getPriceAllAsyncWithCompletableFuture() throws Exception {
	List<String> stores = List.of("storeA", "storeB", "storeC");
	
	List<CompletableFuture<Integer> futures = stores.stream()
		.map(store -> CompletableFuture.supplyAsync(() -> itemRepository.getPriceByStoreName(store)))
		.toList();
		
	Thread.sleep(1000);
	
	futures.stream()
		.map(CompletableFuture::join)
		.forEach(price -> log.info("price = {}", price));
}

해당 코드를 실행시켜보면 이전 parallelStream을 사용한 코드보다 빠른 수행시간을 보여줍니다.

2sec 465ms -> 1sec 457ms

이런 차이가 나는 이유는 parallelStream은 병렬로 실행되긴 하지만, 스트림의 모든 요소가 처리될 때까지 기다려야 합니다. 즉, 모든 요소가 완료될 때까지 블록됩니다. 반면, CompletableFuture는 I/O와 같은 시간이 걸리는 작업을 비동기적으로 요청한 후, 다른 작업을 수행하다가 필요한 시점에 결과를 조회할 수 있습니다. 

 

이러한 이유로 parallelStream은 데이터의 컬렉션 처리, 필터링, 변환과 같은 병렬 데이터 처리 작업에 적합하고, CompletableFuture는 비동기 작업, 네트워크 호출과 같은 비동기 작업의 조합이 필요할 때 적합하다는 것을 알 수 있습니다. 

 

결론

• ForkJoinPool은 Work-Stealing 매커니즘을 사용하는 스레드 풀로 CPU 집약적인 작업을 할 때 최적의 성능을 낼 수 있습니다. 
• 이는 CompletableFuture와 parallelStream에 ForkJoinPool.commonPool()로 전역적으로 제공되어 문제를 발생시킬 수 있어 요청마다 스레드 풀을 만들어 사용하는 방식을 사용하는 것이 좋습니다. 
• CompletableFuture와 parallelStream은 작업을 병렬로 실행 시킬수 있지만, CompletableFuture는 I/O 바운드 작업을 parallelStream은 CPU 바운드 작업에 사용하는 것이 효율적이기에 작업의 특성과 요구사항에 따라 적절히 선택하는 것이 좋습니다.