Java Concurrency Evolution (동시성 처리의 여러가지 방식)

자바 스레드

JVM 스레드와 OS 스레드는 1:1로 맵핑된다.
OS는 범용적이면서 느린 스케줄링을 사용하며, JVM 내부에 대해 아무것도 알지 못한다.
스레드를 만들고 스레드를 오가는 작업은 커널을 거쳐야 하므로 느리다.
OS의 continuation 구현체는 JAVA의 콜 스택 뿐만 아니라 네이비브 콜 스택도 포함하며, 자원을 많이 사용한다.
OS 스레드 개수를 CPU 코어 숫자에 의해 제약받는다.
스레드에 사용되는 스택 메모리는 OS에 의해 힙 외의 영역에 마련된다.

태스트

task는 동시에 호출되어 실행된다.

서비스 A가 호출되면 완료될 때까지 1000ms가 필요하다.
서비스 B가 호출되면 완료될 때까지 500ms가 필요하다.
서비스 A, B의 결과는 파일, DB, S3 등 저장 방식별로 Z회 저장되며, 한 번 저장하는데 300ms가 필요하다. 현실에서는 저장 방식 별로 소요 시간이 다르겠지만 계산의 편의를 위해 같다고 가정한다.
동시 실행이 전혀 없는 No Concurrency 방식에서는 요청 갯수 * (서비스A 처리 시간 + 서비스B 처리 시간 + 저장 횟수 * 저장 시간)만큼의 시간이 필요하다.

반면에 모든 요청이 동시에 실행되는 이상적인 Full Concurrency 방식에서는 Max(서비스A 처리 시간, 서비스B 처리 시간) + 저장 시간 만큼, 그러니까 1,300ms가 필요하다.

동시성 미사용

가장 단순한 방식
자원을 효율적으로 사용하지 못해 성능은 떨어진다.
사용된 유일한 JVM스레드는 하나의 OS스레드를 사용

하나의 OS 스레드는 하나의 CPU 코어를 사용하므로, 나머지 코어는 모두 ㄴ로게 된다.

public void shouldBeNotConcurrent() {

for (int user = 1; user <= USERS; user++) {
    String serviceA = serviceA(user);
    String serviceB = serviceB(user);
    for (int i = 1; i <= PERSISTENCE_FORK_FACTOR; i++) {
        persistence(i, serviceA, serviceB);
    }
}
}

네이티브 멀티 스레딩

멀티 스레딩을 위한 난관

CPU 코어나 메모리 같은 ㅏㅈ원의 효율적 이용
세밀한 스레드 개수 조절이나 관리
제어 흐름과 컨텍스트 유실
실행 동기화
디버깅과 테스트

public void shouldExecuteIterationsConcurrently() throws InterruptedException {

    List<Thread> threads = new ArrayList<>();
    for (int user = 1; user <= USERS; user++) {
        Thread thread = new Thread(new UserFlow(user));
        thread.start();
        threads.add(thread);
    }
    // 종료 조건 - 가장 효율적인 방법은 아니지만 의도대로 동작한다.
    for (Thread thread : threads) {
        thread.join();
    }
}

static class UserFlow implements Runnable {

    private final int user;
    private final List<String> serviceResult = new ArrayList<>();

    UserFlow(int user) {
        this.user = user;
    }

    @SneakyThrows
    @Override
    public void run() {
        Thread threadA = new Thread(new Service(this, "A", SERVICE_A_LATENCY, user));
        Thread threadB = new Thread(new Service(this, "B", SERVICE_B_LATENCY, user));
        threadA.start();
        threadB.start();
        threadA.join();
        threadB.join();

        List<Thread> threads = new ArrayList<>();
        for (int i = 1; i <= PERSISTENCE_FORK_FACTOR; i++) {
            Thread thread = new Thread(new Persistence(i, serviceResult.get(0), serviceResult.get(1)));
            thread.start();
            threads.add(thread);
        }

        // 종료 조건 - 가장 효율적인 방법은 아니지만 의도대로 동작한다.
        for (Thread thread : threads) {
            thread.join();
        }
    }

    public synchronized void addToResult(String result) {
        serviceResult.add(result);
    }
}
// Service와 Persistence 구현 코드는 생략

코드량이 확연히 늘어났다. 수동으로 스레드를 생성하고 제어하고 동기화도 필요하다.
실행 성능은 좋다.
스레드를 무한정 생성할 수는 없다. OS akek ekfmek.

Java 5에 도입된 ExecutorService

스레드 풀링을 통해 새 스레드 생성 부담을 던다.
스레드를 로우 레벨로 다루는 부담을 덜어내는 것이 주 목표다.
태스크는 ExecutorService에 submit되고, 큐에 들어간다.
작업 가능한 스레드가 큐에서 태스크를 가져가서 실행한다.
JVM 스레드 개수가 여전히 OS 스레드 개수에 제한을 받는다.
스레드 풀에서 스레드를 하나 가져가면, 그 스레드는 연산을 수행하지 않더라도 다른 곳에 사용되지 못하도 낭비된다.

public void shouldExecuteIterationsConcurrently() throws InterruptedException {

    for (int user = 1; user <= USERS; user++) {
        executor.execute(new UserFlow(user));
    }

    // 종료 조건
    latch.await();
    executor.shutdown();
    executor.awaitTermination(60, TimeUnit.SECONDS);
}

static class UserFlow implements Runnable {

    private final int user;

    UserFlow(int user) {
        this.user = user;
    }

    @SneakyThrows
    @Override
    public void run() {
        Future<String> serviceA = executor.submit(new Service("A", SERVICE_A_LATENCY, user));
        Future<String> serviceB = executor.submit(new Service("B", SERVICE_B_LATENCY, user));

        for (int i = 1; i <= PERSISTENCE_FORK_FACTOR; i++) {
            executor.execute(new Persistence(i, serviceA.get(), serviceB.get()));
        }

        latch.countDown();
    }
}
// Service와 Persistence 구현 코드는 생략

스레드 풀 크기를 제대로 설정하지 않으면 데드락이 발생하기도 한다.

Fork/Join 프레임 워크

Java 7부터 ExecutorService를 기반으로 만들어져 도입된다.
재귀적으로 더 작은 크기로 쪼갤 수 있는 태스크를 효율적으로 처리하기 위해 만들어짐
ExecutorService를 아직 대체할 수는 없음
Work Stealing

르세드 풀에 있던 A 스레드에 과부하가 걸려서 내부 큐가 꽉 차있을 때, 스레드 풀에 있는 다른 스레드 B가 ExecutorService의 메인 큐에 있는 태스크를 가져오는 대신에, 과부하 걸린 스레드 A 내부 큐에 있는 태스크를 가져와서 처리한다.


public void shouldExecuteIterationsConcurrently() throws InterruptedException {

commonPool.submit(new UserFlowRecursiveAction(IntStream.rangeClosed(1, USERS)
        .boxed()
        .collect(Collectors.toList())));

// Stop Condition
commonPool.shutdown();
commonPool.awaitTermination(60, TimeUnit.SECONDS);
}

public static class UserFlowRecursiveAction extends RecursiveAction {

private final List<Integer> workload;

public UserFlowRecursiveAction(List<Integer> workload) {
    this.workload = workload;
}

@Override
protected void compute() {

    if (workload.size() > 1) {
        commonPool.submit(new UserFlowRecursiveAction(workload.subList(1, workload.size())));
    }

    int user = workload.get(0);

    ForkJoinTask<String> taskA = commonPool.submit(() -> service("A", SERVICE_A_LATENCY, user));
    ForkJoinTask<String> taskB = commonPool.submit(() -> service("B", SERVICE_B_LATENCY, user));

    IntStream.rangeClosed(1, PERSISTENCE_FORK_FACTOR)
            .forEach(i -> commonPool.submit(() -> persistence(i, taskA.join(), taskB.join())));
}

}



## CompletableFuture  

* Java 8에서 도입되어 Fork/Join 프레임워크를 기반으로 만들어져 제공됨 
* 더 개선된 함수형 프로그래밍 스타일 도입
* 로직을 조립하고 결과를 모아서 처리하고 비동기 연산과정을 실행하고 에러를 처리할 수 있는 50여개의 메소드 추가

## Reactive  

* 비동기 데이터 스트림을 처리하고 에러 처리와 배압을 확고하게 지원하는 리액티브 프로그래밍만을 다룬다.  
* 데이터를 발생시키는 Observable, 데이터를 소비하는 Observer, 스레드를 관리하는 Scheduler의 삼위 일체  
* 리액티브 방식을 도입하면 프로그램 흐름 전부가 리액티브 방식으로 같이 바뀌어야 한다는 점에서 전염성이 강하다. 일부에 블로킹 코드가 남아 있으면 리액티브의 장점은 전혀 발휘되지 못한다.  
* 리액티브 구현체도 여러가지가 있다. 처음에는 RxJava가 있었지만 최근에는 스프링의 Reactor가 대세다. 액터 모델을 구현하는 Akka 프레임워크는 RxJava나 Reactor보다 더 급진적인 리액티브 프로그래밍을 적용하고 있다.  

## Project Loom  
* 정식 출시 되지는 않음
* 가상 스레드의 부활  
* JVM 스레드 : OS 스레드 = 1:1 드잇ㄱ이 더이상 성립되지 않음 
* 기존의 JVM 스레드에 비해 훨씬 가볍고 저렴하다.  
* 메타데이터, 스택메모리, 컨택스트 스위치 시간이 네이티브 OS 스레드의 수분의 일밖에 되지 않는다. 
* 아직 지원도구는 충분하지 않다. 
* JVM 차원에서의 개선이기 때문에, 많은 레거시들이 별다른 수정 없이도 성능 개선 효과를 그대로 누릴 수 있다.

yeoseon / tip-archive

[Java] Java Concurrency Evolution (동시성 처리의 여러가지 방식) #267