동시성 심화 — Fork-Join부터 Structured Concurrency까지

public class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;
    private final long[] array;
    private final int start, end;
    // 생성자 생략

    @Override
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) sum += array[i];
            return sum;
        }
        int mid = (start + end) / 2;
        SumTask left = new SumTask(array, start, mid);
        SumTask right = new SumTask(array, mid, end);

        left.fork();                        // 왼쪽을 다른 스레드에 넘김
        long rightResult = right.compute(); // 오른쪽은 현재 스레드에서 계산
        long leftResult = left.join();      // 왼쪽 결과 대기
        return leftResult + rightResult;
    }
}

public class Point {
    private double x, y;
    private final StampedLock lock = new StampedLock();

    public void move(double deltaX, double deltaY) {
        long stamp = lock.writeLock();
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            lock.unlockWrite(stamp);
        }
    }

    public double distanceFromOrigin() {
        long stamp = lock.tryOptimisticRead(); // 락 없이 stamp만 받음
        double currentX = x, currentY = y;

        if (!lock.validate(stamp)) {           // 쓰기가 끼어들었는지 확인
            stamp = lock.readLock();           // 비관적 읽기 락으로 전환
            try {
                currentX = x;
                currentY = y;
            } finally {
                lock.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }
}

// 3단계 파이프라인: 각 단계마다 모든 참여자가 완료해야 다음으로
Phaser phaser = new Phaser(3); // 초기 참여자 3명

for (int i = 0; i < 3; i++) {
    final int workerId = i;
    new Thread(() -> {
        // 페이즈 0: 데이터 로딩
        System.out.println("워커 " + workerId + ": 데이터 로딩 완료");
        phaser.arriveAndAwaitAdvance(); // 모두 도착할 때까지 대기

        // 페이즈 1: 데이터 처리
        System.out.println("워커 " + workerId + ": 데이터 처리 완료");
        phaser.arriveAndAwaitAdvance();

        // 페이즈 2: 결과 저장
        System.out.println("워커 " + workerId + ": 결과 저장 완료");
        phaser.arriveAndDeregister(); // 작업 끝, 참여 해제
    }).start();
}

Phaser phaser = new Phaser(1); // 메인 스레드가 초기 참여자

// 새 참여자 등록
phaser.register(); // 참여자 수: 2
new Thread(() -> {
    // 작업 수행
    phaser.arriveAndDeregister(); // 작업 끝나면 빠지기
}).start();

// 조건에 따라 추가 참여자를 넣을 수도 있음
if (needMoreWorkers) {
    phaser.register();
    launchNewWorker(phaser);
}

phaser.arriveAndAwaitAdvance(); // 메인 스레드도 대기

// 문제가 있는 기존 코드
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

Future<User> userFuture = executor.submit(() -> findUser(userId));
Future<Order> orderFuture = executor.submit(() -> findOrder(orderId));

// findUser()가 예외를 던지면?
// → findOrder()는 아직 실행 중... 스레드 누수!
User user = userFuture.get();   // 여기서 예외
Order order = orderFuture.get(); // 이 줄은 실행도 안 되지만 orderFuture는 아직 돌아가는 중

// Structured Concurrency 방식
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    // 하위 태스크들을 스코프 안에서 실행
    Subtask<User> userTask = scope.fork(() -> findUser(userId));
    Subtask<Order> orderTask = scope.fork(() -> findOrder(orderId));

    scope.join();           // 모든 하위 태스크 완료 대기
    scope.throwIfFailed();  // 하나라도 실패했으면 예외 발생

    // 여기 도달하면 둘 다 성공한 것
    return new Response(userTask.get(), orderTask.get());
}
// try 블록을 벗어나면 아직 실행 중인 하위 태스크는 자동 취소됨

// 여러 미러 서버 중 가장 빨리 응답하는 결과 사용
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    scope.fork(() -> fetchFromMirror1());
    scope.fork(() -> fetchFromMirror2());
    scope.fork(() -> fetchFromMirror3());

    scope.join();

    // 가장 먼저 성공한 결과 반환, 나머지는 자동 취소
    return scope.result();
}

// ScopedValue 선언 (static final이 일반적)
private static final ScopedValue<User> CURRENT_USER = ScopedValue.newInstance();

// 값 바인딩 — 스코프 안에서만 유효
ScopedValue.runWhere(CURRENT_USER, authenticatedUser, () -> {
    // 이 블록 안에서는 CURRENT_USER.get()으로 접근 가능
    handleRequest();
});
// 블록을 벗어나면 자동으로 바인딩 해제 — remove() 필요 없음

// 하위 메서드에서 사용
void handleRequest() {
    User user = CURRENT_USER.get(); // 현재 스코프의 값을 읽음
    processOrder(user);
}

private static final ScopedValue<String> REQUEST_ID = ScopedValue.newInstance();

ScopedValue.runWhere(REQUEST_ID, "req-12345", () -> {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        // fork된 하위 태스크에서도 REQUEST_ID.get() 사용 가능
        scope.fork(() -> {
            String id = REQUEST_ID.get(); // "req-12345" — 부모 스코프에서 상속
            return callServiceA(id);
        });
        scope.fork(() -> {
            String id = REQUEST_ID.get(); // 역시 "req-12345"
            return callServiceB(id);
        });

        scope.join();
        scope.throwIfFailed();
    }
});