Java Fork Join Pool 과 비동기 프로그래밍

package com.example.bkjeon.base.services.api.v1.thread;

import lombok.Builder;
import lombok.Getter;

@Getter
public class Item {

    private String name;
    private Integer price;

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

}

package com.example.bkjeon.base.services.api.v1.thread;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.stereotype.Component;

@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();
    }

}

...

int priceA = itemRepository.getPriceByStoreName("storeA");
int priceB = itemRepository.getPriceByStoreName("storeB");
int priceC = itemRepository.getPriceByStoreName("storeC");

log.info("price = {}", priceA);
log.info("price = {}", priceB);
log.info("price = {}", priceC);

// theAccept(...): 결과를 소비하는 콜백함수로, 리턴타입이 없다. CompletableFuture.supplyAsync(...)에서의 작업이 완료된 후 결과를 처리하는 작업을 지정할 때 사용되며 만약 리턴타입이 있는 콜백 함수를 사용하고 싶다면, thenApply(...) 메서드를 사용하면 된다.
// allOf(...): 여러 CompletableFuture를 병렬로 처리할 수 있도록 해주는 메서드이며, 모든 CompletableFuture 의 처리가 완료될 때 까지 대기하며, 완료된 후 CompletableFuture<Void> 를 반환한다. 그러나 allOf 메서드는 모든 Future 의 결과를 결합한 값을 반환하지 않는다. 따라서 개별 CompletableFuture 의 결과를 사용하려면 join() 메서드를 활용해야 한다. join() 을 호출하여 allOf() 의 작업이 완료될 때 까지 블로킹하여 대기할 수 있다.
CompletableFuture<Void> priceA = CompletableFuture.supplyAsync(() -> {
        log.info("ThreadName = {}", Thread.currentThread().getName());
        return itemRepository.getPriceByStoreName("storeA");
    })
    .thenAccept(price -> log.info("price = {}", price));

CompletableFuture<Void> priceB = CompletableFuture.supplyAsync(() -> {
        log.info("ThreadName = {}", Thread.currentThread().getName());
        return itemRepository.getPriceByStoreName("storeB");
    })
    .thenAccept(price -> log.info("price = {}", price));

CompletableFuture<Void> priceC = CompletableFuture.supplyAsync(() -> {
        log.info("ThreadName = {}", Thread.currentThread().getName());
        return itemRepository.getPriceByStoreName("storeC");
    })
    .thenAccept(price -> log.info("price = {}", price));
CompletableFuture.allOf(priceA, priceB, priceC).join();

log.info("price = {}", priceA);
log.info("price = {}", priceB);
log.info("price = {}", priceC);

// new ForkJoinPool(int n) 을 사용 
// 단, Parallel Stream 별로 ForkJoinPool을 인스턴스화하여 사용하면 OOM(OutOfMemoryError)이 발생할 수 있다. default 로 사용되는 Common ForkJoinPool 은 정적이기 때문에 메모리 누수가 발생하지 않지만, Custom 한 ForkJoinPool 객체는 참조 해제되지 않거나, GC로 수집되지 않을 수 있다. 이 문제는 스레드 풀을 명시적으로 종료하는 것이며 사용방법은 하기 코드처럼 forkJoinPool.shutdown(); 로 종료하자.    
ForkJoinPool forkJoinPool = new ForkJoinPool(6);
forkJoinPool.submit(() -> {   
    integerList.parallelStream().forEach((integer) -> { ...})
}
forkJoinPool.shutdown();    

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

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

Thread.sleep(1000); // 추가하면 기존보다 수행시간이 1초 늘어난 것을 확인할 수 있다.

List<Integer> prices = List.of(100, 500, 3000);
prices.forEach(price -> log.info("price = {}", price));    

// 해당 코드처럼 CompletableFuture 를 사용하면 이전 parallelStream 와 다르게 1초대의 빠른 수행시간을 보여준다. 결론적으론 parallelStream 은 병렬로 실행되긴 하지만 스트림의 모든 요소가 처리될 때까지 기다려야 한다. 즉, 모든 요소가 완료될 때까지 block 된다. 반면 CompletableFuture 는 I/O 와 같은 시간이 걸리는 작업을 비동기적으로 요청한 후, 다른 작업을 수행하다가 필요한 시점에 결과를 조회할 수 있다.
List<String> stores = List.of("storeA", "storeB", "storeC");
List<CompletableFuture<Integer>> futures = stores.stream()
    .map(store -> CompletableFuture.supplyAsync(() -> itemRepository.getPriceByStoreName(store)))
    .collect(Collectors.toList());

Thread.sleep(1000);

futures.stream()
    .map(CompletableFuture::join)
    .forEach(price -> log.info("price = {}", price));