[TIL] Java에서의 병렬 처리

 

 

구현하기 바빠서 넘어간 부분들을 조금이라도 정리해보고자 쓰는 TIL 시리즈입니다

 

 

토스트 가게에 3명의 사람이 각각 햄치즈 토스트, 불갈비 토스트, 햄 토스트를 주문했다.

토스트 가게 사장님은 어떻게 해야 가장 효율적으로 빠른 시간 내에 3개의 토스트를 제조할 수 있을까?

하나씩 토스트를 완성하고 나서 손님에게 서빙한 다음, 그 다음 토스트를 만들기 시작하지는 않을 것이다.

대신 식빵 6개를 동시에 불판 위에 올려두고, 각각의 토핑을 동시에 올려서 가장 먼저 완성되는 순서대로 손님들에게 내어줄 것이다.

 

이것이 바로 우리가 일상에서 자연스럽게 사용하는 '병렬 처리'의 개념이다. 서로 독립적인 작업들을 동시에 진행함으로써 전체 작업 시간을 획기적으로 단축시키는 것이다.

 

최근 우리 팀에서 개발한 해변 정보 조회 API가 바로 그런 사례였다. 각 해변마다 필요한 정보를 가져오기 위해 기상청 API를 3개씩 호출해야 했는데, 이를 순차적으로 처리하다 보니 테스트를 해보면서 "너무 느리다"는 반응이 대다수였다.

 

문제를 분석해보니, 서로 독립적인 API 호출들을 마치 토스트를 하나씩 만들듯이 순서대로 기다리며 처리하고 있었다. 해결책은 명확했다. 바로 '병렬 처리'였다. 하지만 구현에 앞서, Java에서 병렬 처리가 어떤 흐름으로 발전해왔는지 이해해보자.

 

 

 

Java 5 이전

Thread

Java에서 가장 기본적인 병렬 처리의 단위는 Thread라고 한다. 또는 일반적으로는 프로그램 내에서 실행되는 작업의 최소 단위라고 알려져 있다. 자바 프로그램은 기본적으로 메인 스레드(main 함수 실행하는 애) 하나만 가지고 시작하나, Thread를 추가로 만들어 실행하면 메인 스레드 외에도 동시에 여러 코드가 병렬적으로 실행된다.

 

메인 스레드에서 작업1과 2를 실행하면 순차적으로 작업 1이 끝난 뒤 작업 2가 시작될 것이다.

그러나 아래 예시 코드와 같이 여러 개의 스레드를 만들면, 두 작업을 동시에 실행시켜서 좀 더 효율적으로 작업을 끝낼 수 있는 것을 알 수 있다. (병렬 처리)

 

 

Thread = 실행 단위 흐름
start() = 새로운 스레드 실행 (실행할 작업을 넘겨주는 것(제출)과 동시에 스레드 실행) -> 병렬 실행 시작
=> 직접적으로 스레드를 만들어서 실행해야 한다 (관리 번거로움)
=> 실무에서는 직접 Thread보다는 ExecutorService, CompletableFuture 같은 고수준 API를 주로 쓴다

public class ThreadExample {
    public static void main(String[] args) {
        // 새로운 스레드 생성 (익명 클래스)
        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("작업 1 - " + i);
                try { Thread.sleep(500); } catch (InterruptedException e) {}
            }
        });

        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("작업 2 - " + i);
                try { Thread.sleep(500); } catch (InterruptedException e) {}
            }
        });

        // 병렬 실행 시작
        t1.start(); // 새로운 스레드에서 작업1 실행 시작
        t2.start(); // 또 다른 스레드에서 작업2 실행 시작

        System.out.println("메인 스레드 종료");
    }
}

 

 

Thread는 다른 블로그에서도 엄청 수준 높게 설명해주고 있으므로, 아래 포스트를 참고하면 깊게 이해할 수 있을 것이다.

 

https://adjh54.tistory.com/167#1)%20%EC%8A%A4%EB%A0%88%EB%93%9C(Thread)-1

[Java] 스레드(Thread) 이해하기 -1 : 구조, 상태, 예시

 

 

 

 

Java 8 이전

 

ExecutorService (스레드 풀)

 

위의 Thread는 직접적으로 스레드를 만들어서 사용하다보니 스레드 개수의 관리도 어렵고 비효율적이라는 단점이 있다. 작업이 많아질수록 새로운 Thread를 계속적으로 생성해야 하기 때문에 성능도 저하되고 메모리도 낭비될 수 있다. 그래서 Java 5부터 등장한 것이 스레드풀 기반의 ExecutorService이다. 

 

ExecutorService에서 달라지는 점은 작업의 제출과 스레드의 실행이 분리된다는 점이다. (앞서 설명했듯이 Thread에서는 둘 다 동시에 실행한다.)

 

작업 제출(Submit) : ExecutorService에 작업만 제출
스레드 실행 : ExecutorService가 작업을 내부에서 자동으로 스레드 풀에서 실행

 

이를 분리함으로써 생기는 이점은 다음과 같다.

• 큐잉: 작업이 많아도 적절히 대기열에서 관리
• 유연한 실행 제어: 언제, 어떤 스레드에서 실행할지 ExecutorService가 결정
• 리소스 최적화: 스레드 풀 크기로 동시 실행 제한

 

스레드 풀을 통해 스레드를 생성하고 관리하는 기능 제공
스레드를 직접 관리하지 않고 스레드 풀에서 꺼내어 사용하는 방식이다.
작업을 큐에 넣으면 스레드 풀이 알아서 할당 -> 효율적이고 안정적이다.

 

// 스레드 4개짜리 풀 생성
ExecutorService executor = Executors.newFixedThreadPool(4);
// 작업 제출 (submit()으로 작업을 넘기면 풀 안에서 스레드가 자동 배분)
executor.submit(() -> System.out.println(Thread.currentThread().getName()));
// 작업 끝나면 풀 종료
executor.shutdown();

 

 

 

 

 

 

Fork/Join Framework (병렬 분할 정복) 

 

Java 7부터 도입된 병렬 처리용 프레임워크이다. 분할/정복이라는 말과 같이 큰 작업을 여러 개의 작업으로 분할(Fork)하여 병렬적으로 실행하고, 후에 결과를 합치는(Join) 방식이다. 빅데이터 처리와 같은 큰 데이터 집합을 재귀적으로 나눌 수 있는 경우에 적합하다.

 

큰 작업을 분할(Fork) → 병렬 실행 → 결과를 합침(Join)
ForkJoinPool 클래스 사용-> ExecutorService의 일종. Fork/Join 작업을 실행하는 스레드풀 (작업을 병렬로 실행하고 관리)
parallelStream()이 내부적으로 사용

 

 

Fork/Join의 핵심 특징으로 Work Stealing이라는 것이 있다. 예를 들어, 어떤 스레드는 할 일이 많은 반면, 다른 스레드는 놀고 있는 상태라면 비효율적이다. 그래서 놀고 있는 상태의(큐가 비어있는) 스레드가 다른 스레드의 큐에서 작업을 스틸해와서 처리해주는 개념이다. 이는 CPU 활용률을 극대화시켜주고, 전체적인 병렬 처리의 효율도 향상된다. 그리고 동적으로 작업을 배분하므로, 스레드 풀에 비해서도 높은 성능을 보인다. 그렇지만 작은 단위로 나눠지지 않으면 오버헤드가 발생하는 문제점이 있다.

 

class SumTask extends RecursiveTask<Integer> {
    private int[] arr;
    private int start, end;
    private static final int THRESHOLD = 10;

    SumTask(int[] arr, int start, int end) {
        this.arr = arr;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        if (end - start <= THRESHOLD) {
            int sum = 0;
            for (int i = start; i < end; i++) sum += arr[i];
            return sum;
        }
        int mid = (start + end) / 2;
        SumTask left = new SumTask(arr, start, mid);
        SumTask right = new SumTask(arr, mid, end);
        left.fork();  // 왼쪽 작업 비동기 실행
        int rightResult = right.compute(); // 오른쪽은 현재 스레드에서 실행
        int leftResult = left.join();      // 왼쪽 결과 기다리기
        return leftResult + rightResult; // 합친 결과 나옴
    }
}

 

 

 

Java 8 이후

 

Java 8은 함수형 프로그래밍과 병렬 처리에 큰 초점을 맞춘 버전이다. 아래와 같은 기능들이 새로 추가되었다. 

 

 - 람다식: () -> {}
 - Stream API: list.stream().filter().map()
 - parallelStream: list.parallelStream().filter().map()
 - CompletableFuture: 비동기 작업 체이닝

 

위 기능들의 등장 배경은 다음과 같다. 기존 Thread/ExecutorService에서는 여러 비동기 작업을 연결하기 복잡한 부분도 있고, 예외 처리 등의 까다로움이 있다는 한계가 있었기 때문이다.

 

 

CompletableFuture

 

Java 8에서 도입된 클래스로, 자바 5 버전의 Future의 성능을 향상시킨 버전이다.  CompletableFuture는 Future + CompletionStage 인터페이스의 구현체이다.

 

등장 배경

기존에는 비동기 작업 A가 끝나면 B를 실행하고, B가 끝나면 C를 실행하는 것이 복잡했다.  CompletableFuture는 이를 파이프라인 방식으로 간단하게 한다. "A 작업 → 결과 변환 → B 작업 → 최종 처리"를 메서드 체이닝처럼 연결할 수 있다.

 

그런데 여기서 의문이 생긴다. 근데 애초에 비동기 작업인데 왜 A가 끝나면 B를 실행하는거지? 

그 이유는 비동기 작업에서도 순서가 필요한 경우가 있기 때문이다.

 

비동기 => 메인 스레드를 블록하지 않는다는 의미 (순서가 의미가 없다는 얘기가 아님)

메인 스레드 블로킹 vs 논블로킹

 

1. 블로킹 (기존 방식)

메인 스레드가 "잠깐, 결과 올 때까지 기다려" 하며 멈춤

 

2. 논블로킹

메인 스레드가 "백그라운드에서 알아서 해, 나는 다른 일 할게" 하고 하던 일 함

 

블로킹은 기다림, 논블로킹은 위임 후 계속 진행

 

 

실제 상황 예시는 다음과 같다.

 // 예: 사용자 프로필 조회
 1. A 작업: 사용자 ID로 사용자 정보 조회 (비동기)
 2. B 작업: 조회된 사용자의 권한 정보 조회 (비동기)
 3. C 작업: 권한에 맞는 메뉴 구성 (비동기)

 

이러한 로직의 경우에는 A가 완료되어야 B를 실행할 수 있다. A의 결과가 B의 입력으로 필요하기 때문이다.

 

CompletableFuture을 사용하면 기존 방식의 복잡한 코드 구성을 깔끔하게 순서대로 처리할 수 있다.

// 기존 방식의 문제 (Java 5-7, ExecutorService + Future)
// - 각 단계마다 get()으로 메인 스레드가 블로킹됨
// - 수동으로 각 Future를 관리해야 함

  ExecutorService executor = Executors.newFixedThreadPool(3);

  try {
      // A 작업 실행 후 블로킹 대기
      Future<User> userFuture = executor.submit(() -> getUser(userId));
      User user = userFuture.get(); // 메인 스레드 블로킹!

      // B 작업 실행 후 블로킹 대기  
      Future<Permission> permFuture = executor.submit(() -> getPermission(user.getId()));
      Permission permission = permFuture.get(); // 또 블로킹!

      // C 작업 실행 후 블로킹 대기
      Future<Menu> menuFuture = executor.submit(() -> buildMenu(permission));
      Menu menu = menuFuture.get(); // 또 블로킹!

      // 최종 처리
      processMenu(menu);

  } catch (ExecutionException | InterruptedException e) {
      // 예외 처리도 복잡
      handleError(e);
  }


//  CompletableFuture 방식 

  // 각 작업은 비동기로 실행되지만, 논리적 의존관계가 있을 때 순서대로 연결함 (메인 스레드는 논블로킹)
  CompletableFuture.supplyAsync(() -> getUser(userId))      // A
      .thenCompose(user -> getPermission(user.getId()))      // B (A 완료 후)
      .thenApply(permission -> buildMenu(permission))        // C (B 완료 후)
      .thenAccept(menu -> processMenu(menu));                // 최종 처리

 

 

CompletableFuture는 미래에 완료될 작업을 현재 시점에 정의하고 조합할 수 있게 해준다.(계획 하는 것과 같다.) 실제 실행은 백그라운드에서 비동기로 이루어지지만, 코드를 순차적이고 읽기 쉽게 작성할 수 있다.

 

체이닝: thenApply(), thenCompose(), thenCombine() 등으로 작업을 연결
예외 처리: handle(), exceptionally() 메서드로 예외 상황 처리

실행 흐름 : callApi() 실행 (비동기) => result 받음 => process(result) 실행 => finalResult 받음 => System.out.println(finalResult) 실행

 

기본 사용 예시는 다음과 같다. 모든 단계는 파이프라인 방식으로 연결되어 있으며, 각 단계는 이전 단계가 완료되어야 실행된다. 이러한  비동기 작업 체이닝과 조합을 통해 복잡한 워크플로우를 구성할 수 있다.

// 별도 스레드(ForkJoinPool.commonPool())에서 callApi() 메서드를 비동기로 실행
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> callApi());

// callApi()가 완료되면 그 결과(result)를 받아서 process(result) 실행
// process() 메서드의 반환값으로 새로운 CompletableFuture 생성

cf.thenApply(result -> process(result))
  .thenAccept(finalResult -> System.out.println(finalResult));
  
// process()가 완료되면 그 결과(finalResult)를 받아서 출력
// thenAccept는 void를 반환 (최종 소비)

 

이 부분도 아래 포스팅에서 이해하기 쉽게 설명해주고 있다.

https://hongchangsub.com/java-completablefuture/

<Java> CompletableFuture는 왜 필요할까?

 

https://kkkapuq.tistory.com/161

[Java] CompletableFuture의 개념과 동작원리, Thread, Future와의 비교

 

 

 

 

ParallelStream

 

위의 CompletableFuture와 같이 Java 8부터 제공되는 기능이다. Stream API의 순차적 데이터 처리를 여러 스레드로 분산하여 동시에 실행하는 방식이다. 단일 스레드로  처리되던 작업을 CPU의 멀티코어를 활용해 병렬로 처리함으로써 성능을 향상시키는 역할을 한다.

 Stream: 데이터를 파이프라인 방식으로 처리하는 API
  - 컬렉션, 배열, I/O 등의 데이터 소스를 함수형으로 처리
  - 원본 데이터를 변경하지 않음 (불변성)
  - 지연 연산 (Lazy Evaluation) 지원

 

ParallelStream (병렬 스트림): 데이터를 여러 청크로 분할하여 다중 스레드에서 동시에 처리하는 스트림이다.

 

Stream API에 .parallel() 붙이면 내부적으로 ForkJoinPool을 사용해 자동으로 병렬 처리
알아서 병렬로 실행되지만, 세부 제어(스레드 개수, 예외 처리 등)는 제한적

 

 

// 1부터 5까지의 숫자로 구성된 리스트 생성
List<Integer> numbers = Arrays.asList(1,2,3,4,5);

// 병렬 스트림 생성 - 여러 스레드에서 동시 처리
numbers.parallelStream()
       .map(n -> {
		     // 현재 처리 중인 숫자와 실행 중인 스레드 이름 출력
             // 병렬 처리로 인해 다양한 스레드 이름이 출력됨
           System.out.println("처리중: " + n + " - " + Thread.currentThread().getName());
	        
           // 각 숫자에 2를 곱하는 변환 작업 (CPU 집약적이라고 가정)
		   return n * 2;
       })
       .toList();

 

수치 계산, 대용량 데이터 처리 등의 작업을 수행할 때 적합하게 사용될 수 있다.

 

 

 

 

정리

• Thread: 직접적인 단일 스레드 관리
• ExecutorService: 스레드 풀 기반 관리
• CompletableFuture : 비동기 작업 체이닝과 조합 지원
  
• Fork/Join Framework : 병렬 분할/정복 구조 지원
• parallelStream(): 내부적으로 ForkJoinPool 사용 → 데이터를 분할하여 다중 스레드에서 동시 처리

 

 

 

참고

 

 

해당 포스팅에서는 Thread에서부터 parallelStream까지의 병렬 처리 발전 흐름을 소개하므로 자세한 내용들은 아래 링크들을 참고하는 것을 추천한다.

 

https://12bme.tistory.com/699

[Java] 개선된 자바 동시성(1) - 병렬 데이터 처리와 성능

 

https://han5ung.tistory.com/27

Java의 병렬 처리를 알아보자 - Parallel Stream(병렬 스트림)

 

https://jihyunhillcs.tistory.com/42#Java8_%EC%9D%B4%EC%A0%84_

(java/spring) 병렬 프로그래밍 - 동시에 일을 처리하는 방법들