CompleteableFuture를 사용해 비동기 요청 처리하기

rin·2020년 7월 4일
JavaSpring boot
0
post-thumbnail

ref. https://pjh3749.tistory.com/280 해당 블로그에서 일반 java 프로젝트로 진행한 예제를 웹 프로젝트로 변경해서 수행해보겠다.

CompleteableFuture나 비동기 요청, 스레드 등에 대한 개념 이야기는 하지 않을 것이다.

전체 코드는 github에서 확인할 수 있다.

준비

domain

gradle기반의 스프링 부트 어플리케이션을 생성해준다.
domain 패키지를 만들고 하위에 Shop 클래스와 RequestType Enum을 추가해준다.

Shop

이 프로젝트는 데이터베이스에 연결하지 않을 것이다.
목표는 CompleteableFuture를 이용하는 방법을 익히는 것이기 때문에 자바 코드 내에서 모든 작업을 수행할 것이다.
따라서 늘 보였던 Entity나 Table같은 어노테이션은 사용하지 않는다.

@Getter
public class Shop {

    private long id;
    private String name;
    private Random random;

    public Shop(long id, String name){
        this.id = id;
        this.name = name;
        random = new Random(name.charAt(0) * name.charAt(1) * name.charAt(2));
    }

}

RequestType

그래도 "웹 어플리케이션"의 형태를 띄고 있기 때문에 requestParam으로 사용할 Enum을 추가하였다.

public enum RequestType {
    async,
    sync
}

Utils

utils라는 패키지를 생성하고 RandomUtil과 TimerUtil을 추가한다.

RandomUtil

랜덤으로 문자열을 생성해주는 메소드를 포함한다.

public class RandomUtil {

    public static String getRandomString(int length) {
        String id = "";
        for (int i = 0; i < length; i++) {
            double dValue = Math.random();
            if (i % 2 == 0) {
                id += (char) ((dValue * 26) + 65);   // 대문자
                continue;
            }
            id += (char) ((dValue * 26) + 97); // 소문자
        }
        return id;
    }

}

TimerUtil

1초간 작업을 중단하는 메소드를 포함한다.
동기식으로 수행하면 delay() 메소드를 호출하는 곳에서 매번 1초씩 딜레이가 생길 것이고, 비동기식으로 수행하면 결과를 얻는 것과 상관없이 작업이 진행될 것이다.

public class TimerUtil {

    public static void delay() {
        int delay = 1000;
        try {
            Thread.sleep(delay);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}

동기식 처리

Service

우선 동기식 처리를 위한 서비스 메소드를 만들것이다.
service 패키지를 만들고 하위에 ShopService 클래스를 생성한다.

ShopService

getSyncPrice 메소드는 동기식으로 price를 계산하여 그 값을 Map에 저장하여 반환하는데, 이 때 위에서 정의한 TimerUtil.delay()를 사용한다.
즉, price를 계산할 때 마다 1초씩 유휴 시간을 갖게되는 것이다.

@Service
public class ShopService {

    public Map<String, Object> getSyncPrice(){
        long beforeTime = System.currentTimeMillis();
        List<Map<Long, Double>> prices = getPrices();
        long afterTime = System.currentTimeMillis();

        Map<String, Object> response = new HashMap<>();
        response.put("prices", prices);
        response.put("executionTime", (afterTime - beforeTime)/1000);

        return response;
    }

    private List<Map<Long, Double>> getPrices() {
        List<Map<Long, Double>> prices = new ArrayList<>();
        for (int i = 0; i < 10; ++i){
            Shop shop = new Shop(i, RandomUtil.getRandomString(5));
            Double price = getSyncPrice(shop, RandomUtil.getRandomString(3));
            prices.add(Map.of(shop.getId(), price));
        }
        return prices;
    }

    private double getSyncPrice(Shop shop, String product){
        return calculatePrice(shop.getRandom(), product);
    }

    private double calculatePrice(Random random, String product){
        TimerUtil.delay();
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }
}

System.currentTimeMillis()을 이용해서 price를 모두 계산하는데 걸리는 시간을 측정한다.

실제로 price를 계산하는 로직은 getPrices() 메소드이며 최종적으로 calculatePrice() 메소드에서 TimerUtil.delay()를 호출해 강제로 딜레이를 유발한다.

Controller

Controller패키지를 생성해 하위에 ShopApiController를 추가해주자.

ShopApiController

컨트롤러는 웹 어플리케이션의 구실을 갖추고 postman을 이용해 서비스를 수행해보려는 수단이므로 굳이 열심히 ^^.. 작성하지는 않을 것이다.

@Controller
public class ShopApiController {

    @Autowired
    private ShopService shopService;

    @GetMapping("/")
    public ResponseEntity<String> connectTest(){
        return ResponseEntity.ok("success connecting");
    }

    @GetMapping("/price")
    public ResponseEntity<Map> getPrice(@RequestParam("type") RequestType type) {
        if (type.equals(RequestType.sync)){
            return ResponseEntity.ok(shopService.getSyncPrice());
        }

        return null;
    }

}

어플리케이션을 실행하고 postman으로 "localhost:8080/"과 "localhost:8080/price?type=async"를 요청해보자.

👉 localhost:8080/

👉 localhost:8080/price

실제로 응답이 올 때 까지 몇 초 정도 시간이 걸리는 것을 느낄 것이다. 반환된 executionTime 값을 보면 10초가 걸렸음을 알 수 있다.

이는 getPrices 내의 for 문이 총 10번 실행되며 각 getSyncPrice 함수에서 1초씩의 딜레이 시간이 있기 때문이다.

비동기식 처리

CompletableFuture를 사용해 비동기 처리를 해보자.

ShopService

핵심인 CompletableFuture를 사용하고 있는 getAsyncPrice 메소드를 먼저보도록하자.

private Future<Double> getAsyncPrice(Shop shop, String product) {
    CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(
        () -> {
            double price = calculatePrice(shop.getRandom(), product);
            futurePrice.complete(price);
        }
    ).start();
    return futurePrice;
}

getSyncPrice 메소드에서 바로 calculatePrice를 호출해 double type의 price를 반환한 것과 다르게, getAsyncPrice는 Future<Double>를 반환하고 있다.

위 로직은 다음 순서로 작동한다.

  1. getAsyncPrice가 호출
  2. CompletableFuture<Double> 타입의 futurePrice 생성
  3. 새로운 스레드를 생성
  4. 3에서 생성한 스레드가 calculatePrice 호출
  5. calculatePrice가 완료되면 결과값을 futurePrice에 저장
  6. futurePrice를 반환

이렇게만 보면 그냥 동기식이랑 뭐가 다를까 싶을 수도 있다. 🤔
하지만 getAsyncPrice는 또 다른 메소드가 호출하고 있으며 호출될 때 마다 "새로운" 스레드를 만들어 job을 start() 시킨다. 즉, 각 스레드가 job을 완료할 때 까지 대기하지 않게된다.

아래의 getPriceFutures는 shop을 생성하고 getAsyncPrice를 호출하는 로직이 포함된 메소드이다.

private List<Future<Double>> getPriceFutures() {
        List<Future<Double>> futures = new ArrayList<>();
        for (int i = 0; i < 10; ++i) {
            Shop shop = new Shop(i, RandomUtil.getRandomString(5));
            futures.add(getAsyncPrice(shop, RandomUtil.getRandomString(3)));
        }
        return futures;
    }

getAsyncPrice가 총 10번 호출되므로 10개의 스레드가 병렬로 수행된다. 각 Future를 리스트에 넣은뒤 반환한다.

최종적으로 얻는 priceFutures는 for문 내에서 get() 함수를 통해 각 Future가 가지고 있는 calculatePrice의 결과값(double price)을 반환한다.

get을 호출하기 전에 비동기적으로 수행할 다른 작업들을 수행할 수 있으며 (여기서는 또다른 CompletableFuture를 생성하는 것) 만약 get을 호출하는 순간까지 Future의 작업이 완료되지 않았다면 락을 건다.

전체 ShopService 코드는 아래와 같다.

@Service
public class ShopService {

    public Map<String, Object> getPrice(RequestType type) throws ExecutionException, InterruptedException {
        long beforeTime = System.currentTimeMillis();
        List<Map<Long, Double>> prices = type.equals(RequestType.sync) ? getPricesSync() : getPricesAsync();
        long afterTime = System.currentTimeMillis();

        Map<String, Object> response = new HashMap<>();
        response.put("prices", prices);
        response.put("executionTime", (afterTime - beforeTime) / 1000);

        return response;
    }

    private List<Map<Long, Double>> getPricesSync() {
        List<Map<Long, Double>> prices = new ArrayList<>();
        for (int i = 0; i < 10; ++i) {
            Shop shop = new Shop(i, RandomUtil.getRandomString(5));
            Double price = getSyncPrice(shop, RandomUtil.getRandomString(3));
            prices.add(Map.of(shop.getId(), price));
        }
        return prices;
    }

    private List<Map<Long, Double>> getPricesAsync() throws ExecutionException, InterruptedException {
        List<Map<Long, Double>> prices = new ArrayList<>();
        List<Future<Double>> priceFutures = getPriceFutures();

        for (int i = 0; i < 10; ++i) {
            prices.add(Map.of((long) i, priceFutures.get(i).get()));
        }

        return prices;
    }

    private List<Future<Double>> getPriceFutures() {
        List<Future<Double>> futures = new ArrayList<>();
        for (int i = 0; i < 10; ++i) {
            Shop shop = new Shop(i, RandomUtil.getRandomString(5));
            futures.add(getAsyncPrice(shop, RandomUtil.getRandomString(3)));
        }
        return futures;
    }

    private Future<Double> getAsyncPrice(Shop shop, String product) {
        CompletableFuture<Double> futurePrice = new CompletableFuture<>();
        new Thread(
                () -> {
                    double price = calculatePrice(shop.getRandom(), product);
                    futurePrice.complete(price);
                }
        ).start();
        return futurePrice;
    }

    private double getSyncPrice(Shop shop, String product) {
        return calculatePrice(shop.getRandom(), product);
    }

    private double calculatePrice(Random random, String product) {
        TimerUtil.delay();
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }
}

ShopApiController

@GetMapping("/price")
public ResponseEntity<Map> getPrice(@RequestParam("type") RequestType type) throws ExecutionException, InterruptedException {
    return ResponseEntity.ok(shopService.getPrice(type));
}

서비스 측에서 RequestType을 판단해 다른 로직을 수행하기 때문에 위와같이 메소드를 변경해준다.

어플리케이션을 시작하고, localhost:8080/price?type=async로 요청을 보내보자

동기식비동기식

수행시간이 10초에서 1초로 감소한 것을 볼 수 있다. 10개의 멀티스레드로 작동했기 때문에 delay 시간인 1초만 걸린 것이다.

profile
rin
🌱 😈💻 🌱
이전 포스트

[번역] Spring Batch

다음 포스트

[Spring] xml configuration을 java configuration으로 변경하기 (1) ApplicationContext

0개의 댓글