멀티스레드와 동기화

import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.LongSupplier;

public class CounterBenchmark {
    static long syncCount;      static final Object lock = new Object();
    static final AtomicLong atomic = new AtomicLong();
    static final LongAdder adder = new LongAdder();
    static long lockCount;      static final ReentrantLock rlock = new ReentrantLock();

    // 같은 "증가" 작업을 4가지 동기화 방식으로 돌리고 시간을 잰다
    static void bench(String name, Runnable inc, LongSupplier result, int threads, int perThread) throws InterruptedException {
        long t0 = System.nanoTime();
        Thread[] ts = new Thread[threads];
        for (int i = 0; i < threads; i++) {
            ts[i] = new Thread(() -> { for (int k = 0; k < perThread; k++) inc.run(); });
            ts[i].start();
        }
        for (Thread t : ts) t.join();
        System.out.printf("%-14s = %,d  (%d ms)%n", name, result.getAsLong(), (System.nanoTime() - t0) / 1_000_000);
    }

    public static void main(String[] args) throws InterruptedException {
        int threads = 8, per = 1_000_000;
        bench("synchronized",  () -> { synchronized (lock) { syncCount++; } },                    () -> syncCount, threads, per);
        bench("AtomicLong",    atomic::incrementAndGet,                                            atomic::get,     threads, per);
        bench("LongAdder",     adder::increment,                                                   adder::sum,      threads, per);
        bench("ReentrantLock", () -> { rlock.lock(); try { lockCount++; } finally { rlock.unlock(); } }, () -> lockCount, threads, per);
    }
}
// 출력 (ms 값은 CPU·코어 수·부하에 따라 다름. 합계는 항상 8,000,000):
// synchronized   = 8,000,000  (1488 ms)
// AtomicLong     = 8,000,000  (874 ms)
// LongAdder      = 8,000,000  (515 ms)
// ReentrantLock  = 8,000,000  (1344 ms)

import java.util.concurrent.*;

public class OrderFanOut {
    record Order(String id, long amount) {}
    record User(String name, String grade) {}
    record Stock(int qty) {}

    static <T> T slow(long ms, T value) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return value;
    }
    static Order fetchOrder(String id) { return slow(300, new Order(id, 45_000)); }
    static User  fetchUser(String id)  { return slow(300, new User("kim", "VIP")); }
    static Stock fetchStock(String id) {
        if (id.endsWith("9")) throw new IllegalStateException("재고 서비스 다운");
        return slow(300, new Stock(3));
    }

    // 세 조회를 동시에 시작하고, 재고 조회만 실패해도 나머지 결과는 살린다
    static String summary(String orderId, ExecutorService pool) {
        CompletableFuture<Order> orderF = CompletableFuture.supplyAsync(() -> fetchOrder(orderId), pool);
        CompletableFuture<User>  userF  = CompletableFuture.supplyAsync(() -> fetchUser(orderId), pool);
        CompletableFuture<Stock> stockF = CompletableFuture.supplyAsync(() -> fetchStock(orderId), pool)
                .exceptionally(ex -> new Stock(-1));                          // 부분 실패 → fallback
        return orderF.thenCombine(userF, (o, u) -> u.name() + "(" + u.grade() + ") " + o.id() + " " + o.amount() + "원")
                     .thenCombine(stockF, (s, st) -> s + " 재고=" + (st.qty() < 0 ? "조회실패" : st.qty()))
                     .join();
    }

    public static void main(String[] args) {
        try (ExecutorService pool = Executors.newFixedThreadPool(6)) {
            long t0 = System.currentTimeMillis();
            String seq = fetchUser("ORD-1").name() + " " + fetchOrder("ORD-1").amount() + " " + fetchStock("ORD-1").qty();
            long seqMs = System.currentTimeMillis() - t0;

            t0 = System.currentTimeMillis();
            String par = summary("ORD-1", pool);
            long parMs = System.currentTimeMillis() - t0;

            System.out.println("순차: " + seq + " (~" + seqMs / 100 * 100 + "ms)");
            System.out.println("병렬: " + par + " (~" + parMs / 100 * 100 + "ms)");
            System.out.println("부분 실패: " + summary("ORD-9", pool));
        }
    }
}
// 출력 (시간은 환경에 따라 다름. 순차 ≈ 300ms × 3, 병렬 ≈ 300ms × 1 의 비율이 핵심. 아래는 부하가 높은 PC 의 실측):
// 순차: kim 45000 3 (~3800ms)
// 병렬: kim(VIP) ORD-1 45000원 재고=3 (~1200ms)
// 부분 실패: kim(VIP) ORD-9 45000원 재고=조회실패

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class GracefulShutdown {
    static final AtomicInteger goodDone = new AtomicInteger(), badDone = new AtomicInteger();

    // 올바른 워커: 인터럽트 플래그를 확인하고, 예외를 받으면 플래그를 복원한 뒤 빠져나온다
    static Runnable goodWorker(BlockingQueue<Integer> q) {
        return () -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    Integer job = q.poll(50, TimeUnit.MILLISECONDS);   // 대기 중 인터럽트 → InterruptedException
                    if (job == null) continue;
                    Thread.sleep(5);                                    // 처리 (블로킹 구간)
                    goodDone.incrementAndGet();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();                     // 플래그 복원: 호출자도 알 수 있게
            }
            System.out.println(Thread.currentThread().getName() + " 종료, interrupted=" + Thread.currentThread().isInterrupted());
        };
    }

    // 잘못된 워커: InterruptedException 을 삼킨다 → shutdownNow 로도 멈추지 않는다
    static Runnable badWorker(BlockingQueue<Integer> q) {
        return () -> {
            while (true) {
                try {
                    Integer job = q.poll(50, TimeUnit.MILLISECONDS);
                    if (job != null) { Thread.sleep(5); badDone.incrementAndGet(); }
                } catch (InterruptedException e) { /* 삼킴 */ }
            }
        };
    }

    public static void main(String[] args) throws Exception {
        BlockingQueue<Integer> jobs = new LinkedBlockingQueue<>();
        for (int i = 0; i < 10_000; i++) jobs.add(i);

        ExecutorService good = Executors.newFixedThreadPool(2, Thread.ofPlatform().name("good-", 0).factory());
        ExecutorService bad  = Executors.newFixedThreadPool(1, Thread.ofPlatform().name("bad-", 0).daemon(true).factory());
        good.submit(goodWorker(jobs)); good.submit(goodWorker(jobs));
        bad.submit(badWorker(jobs));

        Thread.sleep(200);
        good.shutdownNow();                                             // 워커에 interrupt()
        bad.shutdownNow();
        System.out.println("good 풀 종료됨 = " + good.awaitTermination(1, TimeUnit.SECONDS));
        System.out.println("bad  풀 종료됨 = " + bad.awaitTermination(300, TimeUnit.MILLISECONDS));
        System.out.println("처리 건수 good=" + goodDone.get() + " bad=" + badDone.get());
    }   // bad 워커는 데몬 스레드라 JVM 종료를 막지 못한다. 데몬이 아니었다면 프로그램이 끝나지 않는다
}
// 출력 (good-0/good-1 줄의 순서와 처리 건수는 실행마다 다름):
// good-0 종료, interrupted=true
// good-1 종료, interrupted=true
// good 풀 종료됨 = true
// bad  풀 종료됨 = false
// 처리 건수 good=65 bad=14

import java.util.*;
import java.util.concurrent.*;

public class PhasedSettlement {
    record Tx(int merchant, long amount) {}

    public static void main(String[] args) throws Exception {
        List<Tx> txs = new ArrayList<>();
        Random rnd = new Random(42);
        for (int i = 0; i < 100_000; i++) txs.add(new Tx(rnd.nextInt(5), rnd.nextInt(10_000)));

        int workers = 4, merchants = 5;
        long[][] partial = new long[workers][merchants];              // 워커별 부분합 (공유 쓰기 없음)
        long[] total = new long[merchants];

        CountDownLatch startGate = new CountDownLatch(1);              // 모두 준비되면 동시에 출발
        CyclicBarrier phase = new CyclicBarrier(workers, () -> {       // 4명이 전부 도착하면 딱 한 번 실행
            for (long[] p : partial) for (int m = 0; m < merchants; m++) total[m] += p[m];
            System.out.println("phase 1 완료 → 병합 (barrier action, 한 번만 실행)");
        });
        CountDownLatch done = new CountDownLatch(workers);

        for (int w = 0; w < workers; w++) {
            int id = w;
            new Thread(() -> {
                try {
                    startGate.await();
                    for (int i = id; i < txs.size(); i += workers)      // 파티션: i % workers == id
                        partial[id][txs.get(i).merchant()] += txs.get(i).amount();
                    phase.await();                                      // 전원 집합 후 병합
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    done.countDown();
                }
            }, "worker-" + id).start();
        }
        startGate.countDown();
        done.await();                                                   // await 이후 읽기는 happens-before 로 안전

        for (int w = 0; w < workers; w++)
            System.out.println("worker-" + w + " 부분합 " + Arrays.stream(partial[w]).sum());
        System.out.println("가맹점별 합계 = " + Arrays.toString(total));
        long expected = txs.stream().mapToLong(Tx::amount).sum();
        System.out.println("검산: " + Arrays.stream(total).sum() + " == " + expected + " → " + (Arrays.stream(total).sum() == expected));
    }
}
// 출력:
// phase 1 완료 → 병합 (barrier action, 한 번만 실행)
// worker-0 부분합 124875201
// worker-1 부분합 124983068
// worker-2 부분합 125854664
// worker-3 부분합 125886327
// 가맹점별 합계 = [100086159, 101104212, 99141529, 101151324, 100116036]
// 검산: 501599260 == 501599260 → true

import java.util.*;
import java.util.concurrent.*;

public class VirtualVsPlatform {
    static String call(int id) throws Exception {
        Thread.sleep(50);                                              // 외부 API 지연 (I/O 대기)
        if (id % 250 == 0) throw new IllegalStateException("timeout id=" + id);   // 일부 실패
        return "ok-" + id;
    }

    record Outcome(int ok, List<String> failures, long ms) {}

    static Outcome run(ExecutorService pool, int n) throws Exception {
        long t0 = System.currentTimeMillis();
        List<Future<String>> fs = new ArrayList<>();
        try (pool) {
            for (int i = 1; i <= n; i++) { int id = i; fs.add(pool.submit(() -> call(id))); }
        }                                                              // close(): 전부 끝날 때까지 대기
        int ok = 0;
        List<String> failures = new ArrayList<>();
        for (Future<String> f : fs) {
            try { f.get(); ok++; }
            catch (ExecutionException e) { failures.add(e.getCause().getMessage()); }   // 실패는 모아서 보고
        }
        return new Outcome(ok, failures, System.currentTimeMillis() - t0);
    }

    public static void main(String[] args) throws Exception {
        int n = 1000;
        Outcome p = run(Executors.newFixedThreadPool(16), n);
        Outcome v = run(Executors.newVirtualThreadPerTaskExecutor(), n);
        System.out.printf("플랫폼 풀(16): 성공 %d, 실패 %d, %d ms%n", p.ok(), p.failures().size(), p.ms());
        System.out.printf("가상 스레드  : 성공 %d, 실패 %d, %d ms%n", v.ok(), v.failures().size(), v.ms());
        System.out.println("실패 목록: " + v.failures());
        System.out.println("이론값: 플랫폼 = 1000/16 × 50ms ≈ 3150ms, 가상 ≈ 50ms");
    }
}
// 출력 (ms 는 환경에 따라 다름. 부하가 높은 PC 의 실측이며, 한가한 PC 에서는 이론값에 가까워짐):
// 플랫폼 풀(16): 성공 996, 실패 4, 5696 ms
// 가상 스레드  : 성공 996, 실패 4, 947 ms
// 실패 목록: [timeout id=250, timeout id=500, timeout id=750, timeout id=1000]
// 이론값: 플랫폼 = 1000/16 × 50ms ≈ 3150ms, 가상 ≈ 50ms