Future.get과 병렬 합산

Future를 사용한다고 자동으로 병렬 실행이 되는 것은 아닙니다.

작업 하나를 제출하고 즉시 get한 뒤 다음 작업을 제출하면 호출자가 첫 완료를 기다리는 동안 두 번째 계산은 시작조차 하지 않습니다.

독립 작업을 먼저 제출하고 나중에 결과를 수집하면 실행 구간이 겹칠 수 있습니다.

get이 블로킹하는 위치도 응답 설계의 일부입니다.

메인 요청 스레드에서 무제한 get을 호출하면 느린 작업 하나가 전체 응답을 붙잡습니다.

공통 마감 시간을 전달하거나 비동기로 결과를 연결하고, 시간 초과 뒤의 취소와 종료 정책도 정합니다.

submit·get 반복에 따른 직렬화

import java.util.concurrent.Executors;

public final class SequentialFutureWait {
    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        long start = System.nanoTime();
        try {
            int first = executor.submit(() -> {
                Thread.sleep(200);
                return 20;
            }).get();
            int second = executor.submit(() -> {
                Thread.sleep(200);
                return 22;
            }).get();
            System.out.println("sum=" + (first + second));
            System.out.println("millis=" + (System.nanoTime() - start) / 1_000_000);
        } finally {
            executor.shutdown();
        }
    }
}

한 번의 실행에서는 sum=42, millis=409를 출력했습니다. 두 번째 제출이 첫 get의 반환 뒤에 있으므로 두 sleep 작업은 차례로 실행됩니다. millis에는 제출·수집·출력 비용과 스케줄링이 포함됩니다.

아래 ParallelRangeSum은 sleep 없이 작은 정수 범위를 합산합니다. 이 절에는 같은 sleep 작업을 병렬로 실행한 400ms 대 200ms 비교 실험이 없습니다.

병렬 분할의 검토 항목

• 서로 독립적인 작업만 동시에 제출한다.
• 제출 단계에서 모든 Future 참조를 확보한다.
• 결과 수집 순서와 업무 결과 순서를 별도로 결정한다.
• 분할 비용이 계산 이득보다 작은지 측정한다.
• 한 작업 실패가 전체 실패인지 부분 결과 허용인지 정한다.
• 정한 수집 마감 시간에서 남은 시간을 각 get에 전달한다.

병렬 범위 합산

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

public final class ParallelRangeSum {
    static long sum(long start, long endInclusive) {
        long result = 0;
        for (long value = start; value <= endInclusive; value++) {
            result += value;
        }
        return result;
    }

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        try {
            List<Future<Long>> futures = List.of(
                    executor.submit(() -> sum(1, 50)),
                    executor.submit(() -> sum(51, 100)));
            long total = 0;
            for (Future<Long> future : futures) {
                total += future.get();
            }
            System.out.println("total=" + total);
        } finally {
            executor.shutdown();
        }
    }
}

결과는 5,050입니다.

왼쪽은 1 + … + 50 = 50 × 51 ÷ 2 = 1,275이고, 오른쪽은 51 + … + 100 = (51 + 100) × 50 ÷ 2 = 3,775입니다. 합이 5,050이며 50과 51 사이에 누락·중복이 없습니다. 예컨대 두 번째 범위를 50부터 시작하면 50을 두 번 더해 5,100이 되므로 실행기 선택과 별개로 분할 경계를 먼저 검산해야 합니다.

범위가 겹치거나 빠지지 않는지 분할 경계도 테스트해야 합니다.

이 작은 합산은 구조 설명용입니다. 병렬 준비 비용과 계산 이득은 실제 입력 크기와 부하에서 비교해야 합니다.

이 반복문의 입력 범위도 계약입니다. 예제는 1..100처럼 작고 양수인 범위를 사용합니다. 끝값이 Long.MAX_VALUE이면 마지막 value++가 최솟값으로 넘어가 종료 조건을 다시 만족할 수 있으므로 같은 반복문을 모든 long 범위에 그대로 확대하지 않습니다. 부분합과 전체합이 long 안에 드는지도 별도로 확인해야 합니다. 더 넓은 범위로 확장할 때는 반복 종료 방식과 덧셈 오버플로 처리부터 정합니다.

하나의 기한으로 여러 결과 기다리기

기한의 시작점과 반환값 조건을 먼저 정합니다. 아래 collect의 예산은 메서드 진입 때 시작하므로, 그 전에 작업을 제출하는 데 쓴 시간은 빠집니다. 요청 전체를 같은 기한으로 제한하려면 요청 시작 때 만든 기한에서 남은 시간을 전달하는 별도 설계가 필요합니다. 이 예제에는 나노초로 표현할 수 있는 짧은 양수 예산과 null이 아닌 결과를 전달한다고 가정합니다. Future가 정상적으로 null을 반환할 수는 있지만 마지막 List.copyOf는 null 원소를 거부합니다. 이때도 catch에서 취소를 시도하고 예외를 다시 던지므로, 앞서 얻은 값의 부분 목록을 반환하지 않습니다.

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class DeadlineFutureCollector {
    static <T> List<T> collect(List<Future<T>> futures, Duration budget) throws Exception {
        long deadline = System.nanoTime() + budget.toNanos();
        List<T> results = new ArrayList<>();
        try {
            for (Future<T> future : futures) {
                long remaining = deadline - System.nanoTime();
                if (remaining <= 0) throw new TimeoutException("batch deadline");
                results.add(future.get(remaining, TimeUnit.NANOSECONDS));
            }
            return List.copyOf(results);
        } catch (Exception e) {
            futures.forEach(future -> future.cancel(true));
            throw e;
        }
    }

    public static void main(String[] args) throws Exception {
        var executor = java.util.concurrent.Executors.newFixedThreadPool(2);
        try {
            var futures = List.of(executor.submit(() -> 7), executor.submit(() -> 8));
            System.out.println(collect(futures, Duration.ofSeconds(1)));
        } finally {
            executor.shutdownNow();
        }
    }
}

예시

collect에 들어올 때 공통 예산이 100ms이고 첫 get과 그 사이 처리에 60ms를 썼다고 가정하세요. 다음 get에는 약 40ms가 남습니다. 두 번째 Future의 결과가 이 시점부터 50ms 뒤에 준비된다면 잔여 예산 안에 받지 못합니다. 매번 100ms씩 새로 기다리는 정책과 다릅니다. 이미 끝난 결과도 remaining ≤ 0 검사에 먼저 걸리면 이 코드는 받지 않고 시간 초과를 냅니다. 이 숫자는 원 main의 [7, 8] 실행을 측정한 값이 아닙니다.

결과 수집 방식 비교

분할 수를 정하는 근거

작업 조각이 너무 크면 가장 느린 조각이 전체 완료를 늦추고, 너무 작으면 제출·큐잉·결과 결합 비용이 실제 계산보다 커집니다.

CPU 계산은 사용 가능한 프로세서 수를 출발점으로 삼고, 각 조각의 입력 분포가 비슷한지 측정합니다.

외부 I/O가 섞인다면 프로세서 수만으로 병렬도를 정하지 말고 대상 시스템의 연결 상한과 허용 요청률을 먼저 봅니다.

병렬 합산의 정확성도 성능과 함께 확인합니다.

int 범위를 넘을 수 있으면 부분합과 최종합에 long을 검토하되, long의 범위도 넘는 입력은 별도로 처리해야 합니다. 부동소수점은 결합 순서가 바뀌면 마지막 비트가 달라질 수 있음을 허용 오차에 반영합니다.

기준 순차 구현과 무작위 입력을 비교해 값이 맞는지 확인한 뒤에만 처리 시간을 비교하세요.

입력이 작을 때는 순차 경로를 선택하는 임계값도 둡니다.

측정에서 병렬 준비 비용이 계산 절감보다 큰 구간을 찾아 기준을 정하면 간단한 요청이 큐와 스레드를 불필요하게 거치지 않습니다.

연습 문제

1부터 1,000까지를 두 범위로 나눠 홀수와 짝수 개수를 각각 Callable로 계산하고, 두 작업을 모두 제출한 뒤 결과를 합치세요.

import java.util.concurrent.Executors;

public final class ParallelParitySolution {
    record Counts(int odd, int even) {
        Counts plus(Counts other) { return new Counts(odd + other.odd, even + other.even); }
    }

    static Counts count(int start, int end) {
        int odd = 0, even = 0;
        for (int n = start; n <= end; n++) {
            if ((n & 1) == 0) {
                even++;
            } else {
                odd++;
            }
        }
        return new Counts(odd, even);
    }

    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        try {
            var left = executor.submit(() -> count(1, 500));
            var right = executor.submit(() -> count(501, 1_000));
            System.out.println(left.get().plus(right.get()));
        } finally {
            executor.shutdown();
        }
    }
}

확인 문제

새 범위를 2부터 1,000까지로 두고 2..500과 501..1,000으로 나눠 보세요. 각 부분과 최종 Counts의 odd·even 값은 무엇이며, 전체 개수 999와도 맞나요?

병렬 합산을 채택하는 기준

Future는 작업을 대표할 뿐 병렬 구조를 자동 구성하지 않습니다.

독립 작업을 먼저 모두 제출하고 수집 지점의 블로킹과 시간 상한을 명시해야 실제 병렬성과 응답 규칙이 함께 살아납니다.

범위와 수집 계약을 바꾸는 추가 연습

1. ParallelParitySolution의 대상 범위를 3..999로 바꾸고, 3..500과 501..999를 각각 계산한다. 양 끝은 포함한다. 각 부분과 최종 홀수·짝수 개수, 전체 원소 수를 구하고 경계가 이어지는지 확인하라.
2. 작업 제출 준비에 이미 30ms를 쓴 뒤 collect를 100ms 예산으로 시작한다. 첫 결과와 그 사이 처리에 60ms를 쓰고, 두 번째 결과는 그 시점에서 50ms 뒤에 준비된다. 시간 숫자는 정책 검산을 위한 가정이다. 두 번째 get에 새로 100ms를 주는가? 남은 대기 예산은 얼마이며 결과를 받을 수 있는가? 앞선 30ms는 이 collect 기한 안에 포함되는가?
3. 양수 예산이 충분히 남아 있고 외부 취소·작업 예외가 없다. 이미 정상 완료한 두 Future의 결과가 순서대로 7과 null이다. collect는 [7, null]을 반환하는가? 실패 지점과 재전파되는 예외, 첫 값 7의 부분 반환 여부를 설명하라.