출처1 : https://gurumee92.github.io/2020/09/%EC%9E%90%EB%B0%94%EC%97%90%EC%84%9C%EC%9D%98-%EB%8F%99%EC%8B%9C%EC%84%B1%EA%B3%BC-%EB%B3%91%EB%A0%AC%EC%84%B1/

동시성과 병렬성의 차이

먼저 동시성과 병렬성의 차이가 무엇인지 알아야 한다. 프로그래밍 세계에서 동시성이란, 한 CPU에서 동시에 여러 작업을 하는 것처럼 보이게 만드는 것이다. 다음 그림을 살펴보자.

이렇게 한 CPU에서 2개의 프로세스가 있다고 가정해보자. 이 둘은 엄청나게 짧은 시간에 컨텍스트 스위치가 일어나며 번갈아서 실행된다. 다음 그림처럼 말이다.

그래서 사람 눈으로 볼 때, 동시에 동작하는 것처럼 보인다. 이것이 프로그래밍 세계에서의 동시성이다.

병렬성은 동시성과는 다르게 여러 CPU에서 코드가 병렬적으로 실행되는 것을 말한다. 그림을 살펴보자.

위에 그림은 2개의 CPU 에서 작업이 병렬적으로 실행되는 흐름을 나타내고 있다. 이것들은 한 CPU당 코드가 실행되는 것을 뜻한다. 다음 그림처럼 말이다.

일반적으로 동시성과 병렬성 중 성능은 동시성이 더 우세한다. 왜냐하면, CPU와 하드웨어 등 쓰는 자원이 병렬성에 비해 적기 때문이다.

이제 자바에서 동시성과 병렬성을 어떻게 다루는지 살펴보자. 먼저, 영화 리스트를 제공하는 API에서 데이터를 읽어와 그냥 출력하는 MovieApiFetcher 코드이다. 이 녀석은, Java 11에 만들어진 HttpClient를 사용한다.

다음은 API를 콜하는 코드이다.

package components.fetcher;

import java.net.URI;
import java.net.URISyntaxException;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;

public class MovieApiFetcher {
    private String URL = "https://yts.mx/api/v2/list_movies.json";
    private final static HttpClient httpClient = HttpClient.newBuilder(  )
            .connectTimeout(Duration.ofSeconds(30))
            .build();

    public CompletableFuture<String> fetch(int pageId) {
        URI uri = null;
        try {
            uri = new URI(URL + "?page=" + pageId);
            return httpClient.sendAsync(
                    HttpRequest.newBuilder(uri)
                            .GET()
                            .build(),
                    HttpResponse.BodyHandlers.ofString()).thenApply(HttpResponse::body);
        } catch (URISyntaxException e) {
            e.printStackTrace();
            return new CompletableFuture<>();
        }
    }
}

이제 이를 활용한다.

자바 동시성 코드

먼저 자바에서 동시성을 사용하는 대표적인 수단은 Thread이다. 그리고 여러 스레드를 관리하는 ThreadPool이 있다. 이를 이용할 것이다. 전체 코드는 다음과 같다.

import components.fetcher.MovieApiFetcher;
import org.junit.Test;

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

public class ConcurrentTest {
    @Test
    public void test() {
        long beforeTime = System.currentTimeMillis(); // 시작 시간
        final int nThreads = 50;
        final int requestPerThread = 20;
        final ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
        var fetcher = new MovieApiFetcher();

        for(int thread = 1; thread < nThreads + 1; thread++) {
            final int id = thread;
            Runnable task = () -> {
                for (int i=0; i<requestPerThread; i++) {
                    try {
                        int page = (id + i * nThreads);
                        String result = fetcher.fetch(page).get();
                        System.out.println(page + " " + result);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            executorService.submit(task);
        }

        executorService.shutdown();

        while (!executorService.isTerminated()) { }

        long afterTime = System.currentTimeMillis();
        double secDiffTime = (afterTime - beforeTime)/1000.0;
        System.out.println("전체 테스트 실행시간 : "+ secDiffTime + "초");
    }
}

코드를 하나하나 살펴보자.

long beforeTime = System.currentTimeMillis(); // 시작 시간

// ... 
long afterTime = System.currentTimeMillis();
double secDiffTime = (afterTime - beforeTime)/1000.0;
System.out.println("전체 테스트 실행시간 : "+ secDiffTime + "초");

이 코드는 애플리케이션의 실행 결과를 나타내기 위해 사용하였다.

// ...

final int nThreads = 50;
final int requestPerThread = 20;
final ExecutorService executorService = Executors.newFixedThreadPool(nThreads);

// ...

executorService.shutdown();

while (!executorService.isTerminated()) { }

// ...

위 코드는 ThreadPool을 생성한다. 최대 스레드를 50개, 스레드 내에서 동기적으로 호출되는 API 요청 개수를 20개로 한다. 즉 1000개의 요청이 이루어진다.

또한 요청이 끝나면, ThreadPool에서 모든 쓰레드가 종료될 때까지 기다린다.

var fetcher = new MovieApiFetcher();

for(int thread = 1; thread < nThreads + 1; thread++) {
    final int id = thread;
    Runnable task = () -> {
        for (int i=0; i<requestPerThread; i++) {
            try {
                int page = (id + i * nThreads);
                String result = fetcher.fetch(page).get();
                System.out.println(page + " " + result);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    };
    executorService.submit(task);
}

Runnable 인터페이스인데, ThreadPool에 넣을 때는 Thread가 아닌 Runnable을 넣으는 것이 좋다. Thread 역시 Runnable을 구현하고 있다.

fetcher를 만들고 Runnable 당 20개의 요청을 동기적으로 호출하는 task를 만든다. 그리고 이 태스크를 ThreadPool에 넣어준다.

Callable<T>로 하면 조금 더 성능 향상이 있을 순 있겠으나.. 귀찮아서 Runnable로 짰다.

이게 코드의 다이다. 결과는 어떠한가?

...
전체 테스트 실행시간 : 29.142초

천개 요청하는데 약 30초가 걸렸다.

자바 병렬성 코드

이번엔 병렬성 코드를 보자.

import components.fetcher.MovieApiFetcher;
import org.junit.Test;

import java.util.concurrent.CompletableFuture;
import java.util.stream.IntStream;

public class ParrallelTest {
    @Test
    public void test() {
        long beforeTime = System.currentTimeMillis();
        final int nThreads = 50;
        final int requestPerThread = 20;
        var fetcher = new MovieApiFetcher();

        // 1 ~ 1000 개 변환 과정 int -> Integer -> CompletableFuture<String> -> Void
        IntStream.rangeClosed(1, nThreads * requestPerThread)
                .boxed()
                .map(fetcher::fetch)
                .parallel()
                .map(CompletableFuture::join)
                .forEach(System.out::println);

        long afterTime = System.currentTimeMillis();
        double secDiffTime = (afterTime - beforeTime)/1000.0;
        System.out.println("전체 테스트 실행 시간 : "+ secDiffTime + "초");
    }
}

훨씬 깔끔하긴 하다. Stream API를 이용해 연산을 해서 그런가. 이것도 코드만 살펴보자. 시간 재는 것은 같다. 제일 중요한 코드만 살펴보자.

IntStream.rangeClosed(1, nThreads * requestPerThread)
    .boxed()
    .map(fetcher::fetch)
    .parallel()
    .map(CompletableFuture::join)
    .forEach(System.out::println);

IntStream.rangeClosed는 첫 매개 변수부터, 두 번째 매개 변수까지의 숫자를 만들어낸다. 즉 1 ~ 50 * 20(=1000) 의 숫자 스트림을 만든다.

그 후, 박스 타입인 Stream<Integer>로 변경하고, 이 스트림의 원소들을 다 MovieApiFetcher의 fetch 파라미터로 보낸다.

그 후 CompletableFuture.join을 통해서 API 요청이 끝날 때까지 기다리고 데이터를 출력하는 코드이다. 이 때, join을 하면서 스트림이 동기적인 호출로 변경되기 때문에 성능이 느려진다.

이는 parallel을 사용하여 병렬 스트림으로 스트림을 바꾸면 된다. 그럼 Java가 최적화하여 작업들을 고루게 CPU에 분배시켜준다.

병렬성 코드의 실행 시간은 다음과 같다.

...
전체 테스트 실행시간 : 22.451초

병렬성이 보다 성능이 좋게 나왔다. 왜냐하면 JVM이 Stream API 병렬 연산 시, 적절한 코어 및 스레드 사용을 알아서 조절하기 때문일 것이다. 아마도 Runnable과 CompletableFutre를 조합하는 방법과 쓰레드 수를 조절하면 아마 병렬성보다 빠르게 나올 것으로 기대된다.

병렬성과, 동시성은 비슷하면서도 다르다. 이를 주의하면서 사용하자.

[Java] 동시성과 병렬성 (비동기/async 예제)