CompletableFuture، ForkJoin و موازی‌سازیCompletableFuture, ForkJoin & Parallelism

خب، بیا رک باشیم: همروندی (concurrency) در جاوا جایی است که خیلی از مهندس‌های باتجربه هم گیر می‌کنند — نه چون سخت است، بلکه چون دو ابزار کاملاً متفاوت را با هم قاطی می‌کنند و بعد تعجب می‌کنند چرا برنامه‌ی تولیدی‌شان نصفه‌شب قفل می‌شود. این درس دقیقاً می‌خواهد این گره را باز کند. قرار است از صفر بسازیم، هر واژه‌ی فنی را همان لحظه که می‌بینی معنایش کنیم، و در پایان تو باید بتوانی هر سؤال مصاحبه‌ی این حوزه را با اعتماد جواب بدهی.

بخش ۰ — واژه‌هایی که باید بشناسی

قبل از هر چیز، چند کلمه که مدام تکرار می‌شوند را با یک تصویر ساده جا بیندازیم تا بعداً سردرگم نشوی.

• همروندی (concurrency): یعنی چند کار را طوری مدیریت کنی که انگار همزمان پیش می‌روند. مثل یک آشپز که همزمان چند قابلمه را روی گاز دارد و بینشان می‌چرخد.
• مسدودکننده (blocking): کاری که نخ (thread) را می‌خواباند تا منتظر یک اتفاق بماند — مثلاً منتظر جواب دیتابیس یا شبکه. نخِ مسدودشده هیچ کار دیگری نمی‌کند، فقط منتظر است.
• وابسته به CPU (CPU-bound): کاری که واقعاً پردازنده را مشغول می‌کند، مثل محاسبه‌ی ریاضی سنگین. اینجا نخ نمی‌خوابد، دارد عرق می‌ریزد.
• نخ (thread): یک خط اجرای مستقل. تصورش کن یک کارگر که یک لیست از دستورها را پشت سر هم انجام می‌دهد.
• executor / استخر نخ (thread pool): یک جعبه‌ی از پیش آماده از کارگرها. به‌جای اینکه هر بار کارگر جدید استخدام کنی (گران است)، از استخر یکی را برمی‌داری، کار را می‌دهی، و برمی‌گردانی‌اش.

مدل ذهنی — دو ابزاری که همه اشتباه می‌گیرند

جاوا دو جعبه‌ابزار همروندی دارد که ظاهرشان شبیه است ولی برای دو کار کاملاً مختلف ساخته شده‌اند.

پس دقیق‌تر:

• ForkJoinPool یک موتور کارِ وابسته به CPU (CPU-bound) است. کل طراحی‌اش — صف‌های دوسر (deque) با سرقت کار (work-stealing)، و زمان‌بندی محلی LIFO — بر این فرض بنا شده که وظایف کوتاه، غیرمسدودکننده و به‌صورت بازگشتی قابل تقسیم‌اند. این استخر ستون فقرات parallelStream() و executor پیش‌فرض CompletableFuture است.
• CompletableFuture یک گراف ترکیب ناهمگام (asynchronous composition graph) است؛ یعنی یک promise/future با ترکیب‌کننده‌ها (combinatorها). این کلاس خودش نخ نمی‌سازد و مسدود شدن را مدیریت نمی‌کند؛ فقط تعیین می‌کند callbackهای تو کجا اجرا شوند — و به‌صورت پیش‌فرض این «کجا» همان common pool مربوط به ForkJoin است.

هر جا در ادامه گیج شدی، برگرد به همین جمله. کل درس دارد این یک خط را باز می‌کند.

بخش ۱ — CompletableFuture

ساختن future

قبل از اینکه بتوانی کارها را به هم بچسبانی، باید یک future بسازی. چند راه داریم:

// supplier را روی common pool اجرا می‌کند و بلافاصله برمی‌گردد
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> fetchUser());

// به‌جای آن روی executor داده‌شده اجرا می‌شود
ExecutorService io = Executors.newFixedThreadPool(16);
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> fetchUser(), io);

// مقادیر از پیش تکمیل‌شده (هیچ کار ناهمگامی ندارد)
CompletableFuture<Integer> done = CompletableFuture.completedFuture(42);

// futureای که خودت دستی کاملش می‌کنی — الگوی «promise»
CompletableFuture<String> promise = new CompletableFuture<>();
// ... بعداً، از هر نخی:
promise.complete("value");         // یا promise.completeExceptionally(ex);

بیا این‌ها را باز کنیم:

• supplyAsync یعنی «این کار را برو یک جای دیگر انجام بده و به من یک قبضِ رسید بده که بعداً جوابش را بگیرم.» آن «جای دیگر» به‌صورت پیش‌فرض common pool است، مگر اینکه مثل f2 یک executor خودت بدهی.
• completedFuture یعنی «جواب همین الان آماده است، اصلاً کاری برای انجام نیست.» مثل یک قبض که رویش از قبل جوابش نوشته شده.
• الگوی promise: یک CompletableFuture خالی می‌سازی و بعداً از هر نخ دلخواهی با complete(...) پُرش می‌کنی. این وقتی به‌درد می‌خورد که مثلاً منتظر یک callback از یک کتابخانه‌ی قدیمی هستی.

runAsync هم خواهرِ بدون‌خروجیِ supplyAsync است (یک Runnable می‌گیرد و CompletableFuture<Void> می‌دهد) — وقتی فقط می‌خواهی کاری انجام شود و جوابی لازم نداری.

سه خانواده‌ی ترکیب‌کننده‌ها

حالا که یک future داری، می‌خواهی رویش کار کنی: تبدیلش کنی، به کارِ بعدی بچسبانی، یا با یک future دیگر ترکیبش کنی. این متدها را combinator (ترکیب‌کننده) می‌گویند. در سه محور فکر کن:

مهم‌ترین جفتِ این جدول، thenApply در برابر thenCompose است.

به زبان برنامه‌نویسی تابعی: thenApply دقیقاً map است و thenCompose دقیقاً flatMap. قانون ساده: اگر تابعت یک مقدار ساده برمی‌گرداند از thenApply استفاده کن؛ اگر یک future دیگر برمی‌گرداند از thenCompose.

CompletableFuture<Order> pipeline =
    CompletableFuture.supplyAsync(() -> loadCart(userId))     // CF<Cart>
        .thenApply(cart -> priceCart(cart))                   // CF<PricedCart>  (تبدیل همگام)
        .thenCompose(priced -> chargeAsync(priced))           // CF<Payment>     (future برمی‌گرداند -> flatten)
        .thenApply(payment -> new Order(payment));            // CF<Order>

ببین priceCart یک مقدار عادی می‌دهد پس thenApply است؛ اما chargeAsync خودش یک CompletableFuture می‌دهد پس thenCompose است تا جعبه‌ی تودرتو نگیریم.

thenCombine و allOf / anyOf

تا اینجا یک زنجیره‌ی خطی داشتیم. اما گاهی می‌خواهی دو کارِ مستقل را همزمان اجرا کنی و بعد جواب‌هایشان را کنار هم بگذاری. اینجا thenCombine وارد می‌شود:

CompletableFuture<Integer> price = CompletableFuture.supplyAsync(this::fetchPrice);
CompletableFuture<Integer> stock = CompletableFuture.supplyAsync(this::fetchStock);

// هر دو همزمان اجرا می‌شوند؛ وقتی هر دو تمام شدند ترکیب می‌شوند
CompletableFuture<Quote> quote =
    price.thenCombine(stock, (p, s) -> new Quote(p, s));

نکته‌ی طلایی: چون price و stock هر دو با supplyAsync ساخته شده‌اند، همزمان شروع می‌شوند. thenCombine فقط منتظر می‌ماند تا هر دو تمام شوند و بعد تابع را صدا می‌زند. یعنی زمان کل تقریباً برابر با کندترین‌شان است، نه جمعشان.

وقتی نه دو تا بلکه یک لیست از futureها داری، از allOf استفاده می‌کنی:

List<CompletableFuture<Product>> futures = ids.stream()
    .map(id -> CompletableFuture.supplyAsync(() -> fetch(id), io))
    .toList();

CompletableFuture<List<Product>> all =
    CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
        .thenApply(v -> futures.stream()
            .map(CompletableFuture::join)   // امن: اینجا همه از قبل تکمیل شده‌اند
            .toList());

anyOf برعکسِ allOf است: با نتیجه‌ی اولین futureای که تمام شود کامل می‌شود. برای درخواست‌های hedged (یک درخواست را به دو سرور بفرست، هرکدام زودتر جواب داد را بردار) یا timeout مفید است. فقط یک زخم دارد: چون نمی‌داند بین چند نوع مختلف چه نوع مشترکی هست، CompletableFuture<Object> برمی‌گرداند و باید خودت cast کنی.

انتشار خطا: exceptionally، handle، whenComplete

اینجاست که کاندیداهای ارشد از بقیه جدا می‌شوند. اول یک نکته‌ی کلیدی که خیلی‌ها را گیج می‌کند:

CompletableFuture<String> result =
    CompletableFuture.supplyAsync(() -> { throw new IllegalStateException("boom"); })
        .thenApply(s -> s.toUpperCase())          // رد می‌شود — بالادست شکست خورد
        .exceptionally(ex -> "fallback");         // بازیابی: ex یک CompletionException روی ISE است

ببین thenApply اصلاً اجرا نشد چون مرحله‌ی قبلش شکست خورده بود؛ استثنا از رویش پرید و مستقیم رفت سراغ exceptionally. سه ابزار برای مدیریت خطا داریم و فرقشان دقیقاً همان چیزی است که ازت در مصاحبه می‌پرسند:

تله‌های کلیدی که باید حفظ باشی:

• استثنایی که می‌گیری معمولاً یک CompletionException است که علت واقعی را می‌پیچد. برای دیدن علت واقعی حتماً ex.getCause() را صدا بزن.
• whenComplete استثنا را نمی‌بلعد — future بازگشتی همچنان شکست می‌خورد. این برای log و پاکسازی است، نه بازیابی.
• handle روی هر دو مسیر اجرا می‌شود؛ تنها combinatorی است که می‌گذارد شکست را به موفقیت تبدیل کنی و نوع را عوض کنی.
• هر combinator یک نسخه‌ی *Async هم دارد (exceptionallyAsync، handleAsync) اگر لازم باشد بازیابی به نخ دیگری بپرد.

CompletableFuture<Response> safe =
    callServiceAsync()
        .orTimeout(2, TimeUnit.SECONDS)                 // جاوا ۹+: بعد از ۲ ثانیه با TimeoutException شکست
        .handle((resp, ex) -> {
            if (ex != null) {
                log.warn("degraded", ex.getCause());    // بازکردن CompletionException
                return Response.degraded();
            }
            return resp;
        });

کدام executor چه چیزی را اجرا می‌کند — قانونی که همه را زمین می‌زند

حالا رسیدیم به قلبِ همه‌چیز. اگر فقط یک بخش از این درس را قرار است حفظ کنی، همین است. دو قانون ساده که پشتِ درصد بزرگی از باگ‌های همروندی جاوا هستند:

CompletableFuture.supplyAsync(() -> load(), io)   // روی 'io' اجرا می‌شود
    .thenApply(x -> step1(x))                      // روی 'io' (نخ تکمیل‌کننده)
    .thenApplyAsync(x -> step2(x))                 // به ForkJoinPool.commonPool() می‌پرد
    .thenApplyAsync(x -> step3(x), io);            // روی 'io' اجرا می‌شود

خط‌به‌خط بخوانش: load() روی استخر io اجرا می‌شود. حالا step1 یک thenApply غیر-async است، پس طبق قانون ۱ روی همان نخِ io اجرا می‌شود که کار قبلی را تمام کرد. بعد step2 یک thenApplyAsync بدون executor است، پس طبق قانون ۲ می‌پرد روی common pool. و step3 یک thenApplyAsync با executor io است، پس برمی‌گردد روی io.

پیامدهایی که در تولید گاز می‌گیرند:

• اگر supplyAsync(..., io) کامل شود، آنگاه step1 هم یک نخ از io را اشغال می‌کند — یک زنجیره‌ی طولانی thenApply می‌تواند استخر IO تو را قحطی‌زده کند، حتی اگر فقط یک مرحله «async» بوده باشد.
• اگر future از قبل تمام شده باشد (مثلاً completedFuture)، thenApply تو به‌صورت inline روی نخِ فراخوان اجرا می‌شود — گاهی در تست‌ها غافلگیرکننده است، چون فکر می‌کنی کار در پس‌زمینه می‌رود ولی همان‌جا سرِ جای خودت اجرا می‌شود.
• چون پیش‌فرض common pool است، هر فراخوان مسدودکننده درون یک combinator پیش‌فرض، یک worker از common pool را مسدود می‌کند، و به‌صورت پیش‌فرض تنها Runtime.availableProcessors() - 1 تای آن‌ها وجود دارد. اگر به‌اندازه‌ی کافی مسدود کنی، کل ماشینِ موازیِ JVM (streamها، futureهای دیگر) متوقف می‌شود. همیشه برای کار مسدودکننده یک executor صریح پاس بده.

یک مثال واقع‌گرایانه از ترکیب

بیا همه‌چیز را کنار هم بگذاریم: سه سرویس را همزمان صدا بزن، جواب‌هایشان را ترکیب کن، هرکدام timeout خودش را داشته باشد، و اگر چیزی خراب شد به یک نسخه‌ی تنزل‌یافته برگرد.

// فراخوان همزمان ۳ سرویس، ترکیب، با timeout هر فراخوان و یک fallback سراسری.
CompletableFuture<Profile> profile = supplyAsync(() -> userSvc.get(id), io)
        .orTimeout(500, MILLISECONDS);
CompletableFuture<List<Order>> orders = supplyAsync(() -> orderSvc.recent(id), io)
        .exceptionally(ex -> List.of());               // سفارش‌ها اختیاری‌اند
CompletableFuture<Score> score = supplyAsync(() -> riskSvc.score(id), io)
        .orTimeout(300, MILLISECONDS);

CompletableFuture<Dashboard> dashboard =
    profile.thenCombine(orders, Dashboard::new)        // (Profile, List<Order>) -> Dashboard
           .thenCombine(score, Dashboard::withScore)   // افزودن نتیجه‌ی سوم
           .exceptionally(ex -> Dashboard.EMPTY);      // اگر profile/score شکست خورد، تنزل بده

Dashboard d = dashboard.join();   // یک بار در لبه، فراخوان را مسدود کن

دقت کن به فلسفه‌اش: هر سرویس روی io (استخر اختصاصی، نه common pool) اجرا می‌شود، هرکدام timeout دارد، orders که «اختیاری» است اگر شکست بخورد به لیست خالی برمی‌گردد، و در انتها فقط یک بار، در لبه‌ی گراف، با join() مسدود می‌شویم. درونِ گراف کاملاً غیرمسدودکننده است.

بخش ۲ — ForkJoinPool و سرقت کار (work-stealing)

حالا برویم سراغ آن آشپزخانه. ForkJoinPool برای یک نوع خاص از موازی‌سازی ساخته شده: تقسیم‌وحل (divide-and-conquer).

مدل

ایده ساده است: یک وظیفه‌ی بزرگ را به دو نصفه تقسیم کن (fork)، هر نصفه را جداگانه حل کن، بعد جواب‌ها را با هم ادغام کن (join). و اگر نصفه هنوز بزرگ بود، خودش را باز نصف کن — تا وقتی به تکه‌های کوچکِ قابل‌حل برسیم.

قلبِ فنیِ ForkJoin در نحوه‌ی توزیع کار بین نخ‌هاست. هر نخِ worker یک صف دوسر (deque) دارد — صفی که هم از سر و هم از دُم می‌شود کار برداشت.

پس مالک از سرِ deque به‌صورت LIFO push/pop می‌کند (محلیّت کش خوب، کارهای اخیر داغ‌اند)، و workerهای بیکار از دُمِ deque یک قربانی به‌صورت FIFO سرقت می‌کنند (قدیمی‌ترین، بزرگ‌ترین، کم‌احتمال‌ترین برای نیاز فوری). این عدم‌تقارنِ LIFO-محلی + FIFO-سرقت همه‌ی هسته‌ها را بدون رقابتِ مرکزی مشغول نگه می‌دارد — نکته‌ی کلیدیِ کل طراحی همین است.

مثال RecursiveTask

بیا این را با کد ببینیم. می‌خواهیم جمعِ یک آرایه‌ی بزرگ را موازی محاسبه کنیم. RecursiveTask<V> را extend می‌کنیم چون جواب برمی‌گرداند (Long):

class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;
    private final long[] arr; private final int lo, hi;

    SumTask(long[] arr, int lo, int hi) { this.arr = arr; this.lo = lo; this.hi = hi; }

    @Override protected Long compute() {
        if (hi - lo <= THRESHOLD) {                 // حالت پایه: مستقیم محاسبه کن
            long s = 0;
            for (int i = lo; i < hi; i++) s += arr[i];
            return s;
        }
        int mid = (lo + hi) >>> 1;
        SumTask left  = new SumTask(arr, lo, mid);
        SumTask right = new SumTask(arr, mid, hi);
        left.fork();                                // چپ را به‌صورت ناهمگام زمان‌بندی کن
        long rightResult = right.compute();         // راست را در همین نخ محاسبه کن (هر دو را fork نکن!)
        long leftResult  = left.join();             // منتظر چپِ دزدیده/صف‌شده بمان
        return leftResult + rightResult;
    }
}

long total = ForkJoinPool.commonPool().invoke(new SumTask(data, 0, data.length));

RecursiveAction هم دقیقاً همین است ولی برای وقتی که جوابی برنمی‌گردانی (نسخه‌ی Void).

انتخابِ THRESHOLD هنر است: اگر خیلی بزرگش کنی، موازی‌سازی کافی نمی‌شود و هسته‌ها بیکار می‌مانند؛ اگر خیلی کوچکش کنی، سربارِ fork/join خودش از کارِ واقعی بیشتر می‌شود. آن را طوری بگذار که هر برگ به‌اندازه‌ی کافی بزرگ باشد که سربار را ناچیز کند، اما به‌اندازه‌ی کافی کوچک باشد که بار خوب متعادل شود.

common pool و درجه‌ی موازی‌اش

حالا یک نکته‌ی بسیار مهم: ForkJoinPool.commonPool() یک تک‌نمونه‌ی (singleton) در سطح کل JVM است. یعنی یک استخرِ مشترک برای کل برنامه.

درجه‌ی موازی‌اش (parallelism) به‌صورت پیش‌فرض Runtime.getRuntime().availableProcessors() - 1 است. یعنی روی یک ماشین ۸ هسته‌ای، ۷ worker داری (چرا منهای یک؟ چون نخِ ثبت‌کننده‌ی کار خودش هم می‌تواند دست به کار شود و کمک کند). قابل تنظیم است:

-Djava.util.concurrent.ForkJoinPool.common.parallelism=32

مسدود شدنِ درست: ManagedBlocker

فرض کن واقعاً چاره‌ای نداری جز اینکه درون یک وظیفه‌ی ForkJoin مسدود شوی. راه‌حلِ رسمی این است که به استخر خبر بدهی، تا بتواند یک نخ جبرانی (compensation thread) بالا بیاورد و درجه‌ی موازی را حفظ کند — یعنی جای آن نخی که خوابیده، یکی دیگر بیدار می‌کند تا هسته‌ها بیکار نمانند:

ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    private String result;
    @Override public boolean block() throws InterruptedException {
        result = blockingRemoteCall();   // فراخوان مسدودکننده‌ی واقعی
        return true;                     // true = تمام شد، نیازی به مسدود شدن دوباره نیست
    }
    @Override public boolean isReleasable() {
        return result != null;
    }
});

parallelStream() روی مسیرهای پر از synchronized/Collections و فراخوان‌های JDBC دقیقاً همان مواردی هستند که مسدود شدنِ ساده‌لوحانه بی‌سروصدا توان عملیاتی (throughput) را افت می‌دهد. ManagedBlocker دریچه‌ی فرارِ مجاز است — اما صادقانه بگویم، روی JDKهای مدرن پاسخِ تمیزتر یک executor اختصاصی یا virtual threadهاست.

بخش ۳ — جزئیات درونی parallel stream

حالا که ForkJoin را می‌شناسی، parallelStream() دیگر جادو نیست. وقتی collection.parallelStream() (یا stream().parallel()) می‌زنی، خط لوله روی همان common pool اجرا می‌شود و منبع را از طریق یک شیء به‌نام Spliterator تقسیم می‌کند.

قواعد سرانگشتی که مستقیم از این جزئیات درونی می‌آیند:

• موازی‌سازی فقط وقتی کمک می‌کند که: منبع یکنواخت و ارزان تقسیم شود (آرایه‌ها، ArrayList، IntStream.range — عالی؛ چون می‌شود وسطشان را با یک محاسبه پیدا کرد. اما LinkedList و منابع مبتنی بر Iterator — ضعیف، چون برای رسیدن به وسط باید از اول قدم بزنی)، کارِ هر عنصر سنگین از نظر CPU باشد، و N بزرگ باشد. یک اکتشافِ تقریبی: N × هزینه‌هرعنصر باید خیلی بیشتر از حدود ۱۰۰ میکروثانیه باشد تا سربارِ موازی‌سازی صرف کند.
• برای reduction، خط لوله باید بدون‌حالت (stateless)، غیرمداخله‌گر (non-interfering) و شرکت‌پذیر (associative) باشد. یعنی: تابعت نباید حالتِ بیرونی نگه دارد یا منبع را وسطِ کار عوض کند، و ترتیبِ گروه‌بندی نباید جواب را عوض کند. reduce باید شرکت‌پذیر باشد؛ collect از یک combiner برای چسباندن نتایج جزئی استفاده می‌کند.
• هرگز در یک parallel stream کار I/O انجام نده — workerهای common pool را مسدود و بقیه را قحطی‌زده می‌کنی. (همان جمله‌ی کلیدی فصل، بار دیگر.)
• forEach هیچ ترتیبی تضمین نمی‌کند؛ اگر به ترتیبِ برخورد (encounter order) نیاز داری از forEachOrdered استفاده کن — که البته هزینه دارد چون باید هماهنگی برقرار کند.

// وابسته به CPU، منبع قابل‌تقسیم، reduction شرکت‌پذیر — انتخاب خوب
long primes = LongStream.rangeClosed(2, 10_000_000)
        .parallel()
        .filter(this::isPrime)
        .count();

این یک نمونه‌ی عالی است: منبع (rangeClosed) کاملاً ارزان نصف می‌شود، کارِ هر عنصر (isPrime) واقعاً CPU-bound است، و N هم بزرگ است.

و یک ترفندِ شناخته‌شده: اگر بخواهی یک parallel streamِ خاص را از common pool فراری بدهی (مثلاً چون نمی‌خواهی بقیه‌ی برنامه را تحت‌تأثیر بگذارد)، آن را درون ForkJoinPool خودت submit کن:

ForkJoinPool custom = new ForkJoinPool(4);
long r = custom.submit(() ->
    list.parallelStream().filter(this::heavy).count()
).join();   // این parallel stream حالا روی 'custom' اجرا می‌شود، نه common pool

بخش ۴ — future در برابر reactive (به‌اختصار)

گاهی در مصاحبه می‌پرسند «فرق CompletableFuture با reactive چیست؟» جوابش کوتاه اما مهم است.

CompletableFuture یک مقدارِ ناهمگامِ تکی را با ترکیب مدل‌سازی می‌کند — «یک جواب، بعداً می‌آید.» اما Reactive Streams (‏Mono/Flux در Project Reactor، یا RxJava) جریان‌هایی از ۰ تا N مقدار با backpressure را مدل می‌کنند و مجموعه‌ی عملگرِ بسیار غنی‌تری دارند (retry، buffer، window، merge، کنترل جریان).

قاعده‌ی انتخاب: از CF برای «این سرویس‌ها را صدا بزن، ترکیب کن، یک‌بار برگردان» استفاده کن؛ از reactive وقتی جریان، backpressure داری یا به ترکیب در طول زمان نیاز داری. Mono در Reactor تقریباً یعنی «CF با صدها عملگر و subscription تنبل.»

بخش ۵ — virtual threadها معادله را عوض می‌کنند (جاوا ۲۱)

حالا بزرگ‌ترین تحول. تا اینجا کلِ فلسفه‌ی CompletableFuture این بود که «نخ‌های سیستم‌عامل گران‌اند، پس روی مسدود شدن هدرشان نده — به‌جایش callback بنویس.» اما virtual threadها (نخ‌های مجازی، JEP 444، پایدار در جاوا ۲۱) این فرض را زیر و رو می‌کنند: مسدود شدن را دوباره ارزان می‌کنند.

با virtual threadها می‌توانی کدِ مسدودکننده‌ی خطی و ساده بنویسی و هزاران تای آن را همزمان اجرا کنی:

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    var f1 = executor.submit(() -> userSvc.get(id));    // فراخوان مسدودکننده، اما روی یک vthread
    var f2 = executor.submit(() -> orderSvc.recent(id));
    return new Dashboard(f1.get(), f2.get());           // ساختارمند، خوانا، بدون زنجیره
}

ببین چقدر خوانا است — بدون هیچ thenCombine و زنجیره‌ای، فقط «این دو کار را همزمان انجام بده و جوابشان را بگیر.»

پس این‌ها رقیب هم نیستند، مکمل‌اند:

• از CompletableFuture برای عملگرهای ترکیبش استفاده کن، وقتی واقعاً به یک گراف نیاز داری (thenCombine، allOf، timeoutها).
• به virtual threadها (یا Structured Concurrency با StructuredTaskScope) روی بیاور وقتی منطق واقعاً «این‌ها را موازی انجام بده و join کن» است.
• ForkJoin را برای موازی‌سازیِ خالصِ CPU-bound نگه دار.

تله‌ها و نکات ظریف

بیا همه‌ی تله‌هایی که در راه دیدیم را یک‌جا جمع کنیم — این‌ها همان‌هایی‌اند که در تولید گاز می‌گیرند:

• مسدود شدن روی common pool. supplyAsync/parallelStream پیش‌فرض + JDBC/HTTP = قحطی. یک Executor اختصاصی پاس بده.
• پیچیده شدن در CompletionException. catch/exceptionally تو wrapper را می‌بیند؛ getCause() را صدا بزن.
• whenComplete بازیابی نیست. دوباره پرتاب می‌کند؛ فقط handle/exceptionally بازیابی می‌کنند.
• allOf مقادیر را از دست می‌دهد. بعد از تکمیلِ allOf، هر future را دوباره join کن.
• combinatorهای غیر-async نخِ تکمیل‌کننده را می‌دزدند. یک زنجیره‌ی طولانی thenApply روی هر نخی که مرحله‌ی قبل را تمام کرده اجرا می‌شود — شاید استخر IO تو، شاید یک worker از common pool.
• فراموش کردن اینکه future مشتاق است. supplyAsync همین حالا شروع می‌شود؛ هیچ subscribe تنبلی وجود ندارد.
• fork کردن هر دو شاخه. یکی را inline محاسبه کن؛ دیگری را fork کن.
• fan-out بی‌کران. نگاشتِ ۱۰۰ هزار آیتم به supplyAsync(..., commonPool) آن را اشباع می‌کند — همروندی را با یک executor اندازه‌گذاری‌شده یا یک semaphore محدود کن.

بهترین روش‌ها

۱. مالکِ executorهای خودت باش برای هر چیزی که مسدود می‌شود. هرگز common pool را برای I/O باور نکن. ۲. نخ‌ها را نام‌گذاری کن (ThreadFactory) تا stack dumpها قابل‌دیباگ باشند. ۳. join() را در لبه‌ی گراف ترجیح بده؛ درونِ گراف را غیرمسدودکننده نگه دار. ۴. به هر فراخوان خارجی orTimeout/completeOnTimeout اضافه کن. ۵. خطاها را با handle/exceptionally مدیریت کن؛ همیشه getCause(). ۶. RecursiveTask را فقط برای تقسیم‌وحلِ وابسته به CPU با آستانه‌ی تنظیم‌شده به‌کار ببر. ۷. روی جاوا ۲۱+، برای fan-out مربوط به I/O، virtual threadها / structured concurrency را ترجیح بده؛ CompletableFuture را برای گراف‌های ترکیبِ واقعی نگه دار؛ ForkJoin را برای موازی‌سازیِ وابسته به CPU.

سؤالات مصاحبه

حالا بخشی که برایش آمده‌ای. هر سؤال را جوری جواب می‌دهیم که هم درست باشد هم نشان‌دهنده‌ی عمق.

var cf = CompletableFuture.supplyAsync(() -> 1)
    .thenApply(x -> x / 0)
    .thenApply(x -> x + 1)
    .exceptionally(ex -> -1);
System.out.println(cf.join());

List<CompletableFuture<Integer>> fs = ...;
CompletableFuture.allOf(fs.toArray(new CompletableFuture[0]));
List<Integer> results = fs.stream().map(CompletableFuture::join).toList();

نکاتِ سنیور و موارد پیشرفته

تا اینجا فصل، ساز و کارِ CompletableFuture، ForkJoin و virtual thread را کامل یادت داد. اما آن‌چه یک سنیور را از یک میدلِ خوب جدا می‌کند، نه دانستنِ این API‌ها، بلکه دانستنِ جاهایی است که این API‌ها بی‌سر و صدا دروغ می‌گویند — کارهایی که فکر می‌کنی انجام می‌دهند ولی نمی‌دهند، و درست همان‌ها هستند که ساعت ۳ بامداد صفحه‌ی pager تو را روشن می‌کنند. این بخش دقیقاً همان لایه است.

۱. کنسل کردنِ یک future یک دروغ است — و orTimeout نشتی می‌سازد

این نکته‌ای است که خودِ Javadoc هم صریح نمی‌گویدش و در تولید خون به پا می‌کند. تصور رایج این است: «اگر روی future صبرم تمام شد، cancel(true) می‌زنم و کارِ پشتش قطع می‌شود.» این غلط است.

CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> slowDbCall(), io);
boolean cancelled = f.cancel(true);   // true = mayInterruptIfRunning

پارامتر mayInterruptIfRunning در CompletableFuture کاملاً نادیده گرفته می‌شود. برخلاف FutureTask، یک CompletableFuture هیچ ارجاعی به نخی که supplyAsync را اجرا می‌کند نگه نمی‌دارد؛ پس نمی‌تواند آن نخ را interrupt کند. تنها کاری که cancel می‌کند این است که خودِ future را به حالتِ استثنایی (CancellationException) می‌برد تا مرحله‌های پایین‌دستِ گراف رها شوند. اما slowDbCall() تا آخر اجرا می‌شود و کانکشنِ دیتابیس تا لحظه‌ی برگشتش checked-out می‌ماند.

همین منطق برای orTimeout هم برقرار است:

CompletableFuture<Response> r = callServiceAsync().orTimeout(2, SECONDS);

نتیجه‌ی سنیوری: کنسل در دنیای CompletableFuture تعاونی (cooperative) نیست مگر این‌که تو دستی سیمش را بکشی. یا کلاینتی به کار بگیر که واقعاً کنسل را propagate کند (مثل HttpClient.sendAsync که futureِ برگشتی‌اش به کنسلِ واقعی سیم‌کشی شده)، یا از StructuredTaskScope استفاده کن که هنگام شکستِ scope، subtask‌ها را واقعاً interrupt می‌کند.

۲. anyOf و allOf برادرها را نمی‌کشند

فصل گفت allOf مقادیر را دور می‌ریزد و anyOf اولین جواب را برمی‌گرداند. اما دو ظرافتِ حیاتی جا ماند:

• anyOf با «اولین چیزی که کامل شود» کامل می‌شود — حتی اگر آن کامل‌شدن یک شکست باشد. یعنی یک شکستِ سریع بر یک موفقیتِ کند برنده می‌شود. اگر داری با anyOf درخواستِ hedged می‌زنی (همان درخواست را به دو سرور بفرست، هرکدام زودتر جواب داد بردار)، یک خطای فوریِ سرورِ اول کلِ نتیجه را شکست می‌دهد در حالی که سرورِ دوم داشت درست جواب می‌داد.
• هیچ‌کدام از دو تا برادرهای بازنده را کنسل نمی‌کنند. بعد از این‌که anyOf برنده را برگرداند، همه‌ی futureهای دیگر تا آخر اجرا می‌شوند و منبع می‌سوزانند. allOf هم به‌محضِ اولین شکست، به‌صورت استثنایی کامل می‌شود ولی بقیه‌ی futureها بی‌مراقب ادامه می‌دهند.

۳. یک future را به بیرون نده مگر محافظت‌شده

وقتی متدِ کتابخانه‌ات CompletableFuture<T> برمی‌گرداند، هر فراخوانی می‌تواند روی آن complete(...), obtrudeValue(...) یا cancel() صدا بزند و کلِ گرافِ پایین‌دستِ تو را بدزدد — انگار یک متغیرِ private را public کرده باشی.

// بد: هر کسی می‌تواند نتیجه‌ی تو را جعل کند
public CompletableFuture<Price> quote() { return internalFuture; }

// خوب: نسخه‌ای که متدهای completion رویش استثنا می‌اندازند
public CompletionStage<Price> quote() { return internalFuture.minimalCompletionStage(); }

minimalCompletionStage() (جاوا ۹) یک CompletionStage می‌دهد که فقط combinatorها را دارد و صدا زدنِ complete/cancel/get رویش UnsupportedOperationException می‌دهد. copy() هم یک کپیِ وابسته می‌دهد که کامل‌کردنش روی اصل اثر ندارد. و obtrudeValue/obtrudeException را فقط در تست به‌کار ببر — این‌ها یک futureِ ازقبل‌کامل‌شده را با زور بازنویسی می‌کنند و همه‌ی invariantها را می‌شکنند.

۴. common pool داخل کانتینر — دامِ Kubernetes

فصل گفت common pool پیش‌فرض availableProcessors() - 1 نخ دارد. اما در دنیای کانتینر این عدد گاز می‌گیرد:

• از JDK 10 (و 8u191) به بعد availableProcessors() نسبت به cgroup آگاه است و به CPU limit احترام می‌گذارد.
• اگر pod تو یک vCPU limit دارد، cores - 1 می‌شود صفر و common pool عملاً به بی‌موازی‌سازی فرو می‌ریزد: parallelStream() روی نخِ صداکننده سریال اجرا می‌شود. آن سرعتی که در لپ‌تاپِ ۸ هسته‌ای دیدی، در prod ناپدید می‌شود.
• برعکس، اگر limit نگذاری، JVM ممکن است همه‌ی هسته‌های نودِ میزبان را ببیند و pool‌ای بسیار بزرگ‌تر از سهمِ واقعی‌ات بسازد.

۵. Pinning در virtual thread — دامِ مدرن و رفعش در جاوا ۲۴

فصل گفت virtual thread بلاک‌شدن را ارزان می‌کند، ولی یک ستاره‌ی مهم دارد که تا جاوا ۲۳ خون‌ریزی می‌کرد: pinning.

وقتی یک virtual thread وارد یک بلاکِ synchronized می‌شد، در نسخه‌های تا جاوا ۲۳ monitorِ آن قفل روی نخِ حاملِ (carrier) پلتفرمی ثبت می‌شد، نه روی خودِ virtual thread. نتیجه: تا وقتی داخلِ آن بلاک بودی، virtual thread به carrier میخکوب (pinned) بود و اگر همان‌جا بلاک می‌شدی (مثلاً I/O داخلِ synchronized)، carrier نمی‌توانست آزاد شود و به virtual threadِ دیگری برسد. اگر همه‌ی carrierها این‌طور میخکوب می‌شدند، کلِ scheduler خفه می‌شد — و در بدترین حالت، deadlock.

۶. ThreadLocal، MDC و ScopedValue

با virtual threadها دو مشکلِ زمینه‌ای (context) بزرگ می‌شود:

• حافظه: اگر میلیون‌ها virtual thread داشته باشی و هرکدام یک ThreadLocal سنگین بگیرد، مصرفِ حافظه منفجر می‌شود. InheritableThreadLocal هم بدتر است چون به هر child کپی می‌شود.
• گم‌شدنِ context در پرش‌های async: مقادیری مثل MDC (همان traceId در لاگ‌ها) و SecurityContext روی ThreadLocal سوارند. وقتی یک thenApplyAsync به نخِ دیگری می‌پرد، آن ThreadLocal همراه نمی‌آید — و یک‌دفعه لاگ‌هایت traceId را گم می‌کنند و امنیتت خالی می‌شود. همین مشکل سرِ مرزِ executor.submit(...) هم هست.

۷. Structured Concurrency — جانشینِ واقعیِ گراف‌های شکننده

مشکلی که CompletableFuture بد حلش می‌کند این است: اگر یکی از N تا subtask شکست بخورد، بقیه به‌صورت یتیم ادامه می‌دهند و منبع می‌سوزانند؛ هیچ کسی مالکِ چرخه‌ی عمرِ همه‌شان نیست. Structured Concurrency دقیقاً همین را حل می‌کند: subtaskها فرزندِ یک scope می‌شوند، و مثل یک بلوکِ try با کروشه، یا همه با هم موفق می‌شوند یا با شکستِ یکی، بقیه کنسل و interrupt می‌شوند.

نمودارِ درختِ scope (هر subtask فرزندِ scope است و در همان بلوک join می‌شود):

flowchart TD
  Scope[StructuredTaskScope] --> A[fork: userSvc.get]
  Scope --> B[fork: orderSvc.recent]
  Scope --> C[fork: riskSvc.score]
  Scope --> J{join}
  J -->|any fails| Cancel[cancel & interrupt siblings]
  J -->|all ok| Combine[combine results]

// نسخه‌ی جاوا ۲۵ (هنوز preview): factory استاتیک open() + یک Joiner برای سیاست
try (var scope = StructuredTaskScope.open(Joiner.<Object>allSuccessfulOrThrow())) {
    var user  = scope.fork(() -> userSvc.get(id));      // فرزندِ scope
    var order = scope.fork(() -> orderSvc.recent(id));
    scope.join();                                        // منتظرِ همه؛ اگر یکی بترکد، بقیه interrupt
    return new Dashboard(user.get(), order.get());
}   // بستنِ scope تضمین می‌کند هیچ subtaskِ یتیمی جا نمی‌ماند

۸. Retry با backoff — جایی که exceptionally کم می‌آورد

exceptionally فقط می‌تواند به یک مقدارِ ساده برگردد؛ نمی‌تواند یک تلاشِ دوباره‌ی async راه بیندازد. برای این، جاوا ۱۲ متدِ exceptionallyCompose را داد که خطا را به یک future دیگر می‌بندد. همراهش delayedExecutor (جاوا ۹) می‌آید که بدونِ Timer جداگانه، یک اجرا را عقب می‌اندازد:

CompletableFuture<Resp> withOneRetry() {
    return callAsync()
        .exceptionallyCompose(ex ->            // خطا؟ یک future تازه بساز (نه یک مقدار)
            CompletableFuture.supplyAsync(this::callAsyncOnce,
                CompletableFuture.delayedExecutor(200, MILLISECONDS)));  // بعد از ۲۰۰ms دوباره
}

۹. تضمینِ حافظه (happens-before) و پرشِ بی‌فایده

دو نکته‌ی ریزِ سنیوری برای بستنِ بحث:

• CompletableFuture یک رابطه‌ی happens-before برقرار می‌کند: هر کاری که در نخِ کامل‌کننده قبل از complete(v) انجام شده، برای مرحله‌های وابسته قابل‌مشاهده است. یعنی برای خواندنِ مقدارِ تولیدشده نیازی به volatile یا قفلِ اضافی نداری. (اما این تضمین فقط شاملِ خودِ آن مقدار است، نه هر state جانبیِ mutable که بین مرحله‌ها به اشتراک گذاشته‌ای.)
• thenApplyAsync(fn, sameExecutor) باز هم می‌پرد: حتی اگر همین الان روی همان executor باشی، نسخه‌ی *Async یک تسکِ جدید submit می‌کند — یعنی صف‌بندی و context switchِ اضافه. اگر به پرش نیاز نداری، combinatorِ غیر-async ارزان‌تر است. پرشِ عمدی فقط وقتی ارزش دارد که بخواهی از یک نخِ گران (مثلاً نخِ event-loop یا IO pool) فرار کنی.

سؤالات مصاحبه‌ی سنیور (سطحِ سخت)