Executor আর CompletableFuture — thread pool, Future আর অপেক্ষাহীন কাজের শিকল

১ · প্রতিটি কাজে নতুন thread কেন ভুল — thread pool-এর ধারণা

পর্ব ৪০-এ আপনি thread নিজে হাতে তৈরি করেছেন: new Thread(task).start(). সেখানে দেখেছেন race condition, synchronized আর thread-safe কোডের প্রয়োজন। এবার প্রশ্ন বদলায়: একটি অনলাইন দোকানের সার্ভারে প্রতি সেকেন্ডে কয়েকশো অনুরোধ আসে। প্রতিটির জন্য কি একটি করে নতুন thread?

নতুন thread দামি কেন? Java-র সাধারণ platform thread আসলে অপারেটিং সিস্টেমের একটি thread. প্রতিটির জন্য OS-কে কাঠামো বানাতে হয়, আর সাধারণত ১ MB পর্যন্ত stack-এর জায়গা সংরক্ষিত থাকে। হাজার হাজার thread একসাথে থাকলে CPU-কে বারবার এক thread থামিয়ে আরেকটি চালাতে হয় — এই context switch-এ সময় যায় আসল কাজের বদলে। আর সীমা না থাকলে একটি ভিড়ের মুহূর্তেই মেমরি ফুরিয়ে OutOfMemoryError: unable to create native thread-এর মতো ভুল আসে। Pool এই তিন সমস্যাই সামলায়: thread একবার তৈরি হয়ে বারবার ব্যবহার হয়, সংখ্যা সীমিত থাকে, আর বাড়তি কাজ সারিতে অপেক্ষা করে।

1var pool = Executors.newFixedThreadPool(3);2for (Task t : tasks)        // ৬টি কাজ3    pool.submit(t);4// worker: সারি থেকে নাও → চালাও → আবার নাও5pool.close();               // সব শেষের অপেক্ষা

২ · Executor, ExecutorService আর Executors-এর কারখানা

নামগুলো কাছাকাছি, তাই শুরুতেই আলাদা করে নিন:

ভেতরের যন্ত্র — ThreadPoolExecutor

প্রথম তিনটি factory আসলে একই class-এর ভিন্ন সেটিং: ThreadPoolExecutor. বড় প্রজেক্টে অনেকে factory বাদ দিয়ে এটাই সরাসরি তৈরি করেন, যাতে প্রতিটি সীমা চোখের সামনে থাকে:

ExecutorService pool = new ThreadPoolExecutor(
    4, 8,                                   // core: সবসময় ৪টি, max: সর্বোচ্চ ৮টি
    30, TimeUnit.SECONDS,                   // বাড়তি thread কতক্ষণ খালি বসে থাকবে
    new ArrayBlockingQueue<>(100),          // সারিতে সর্বোচ্চ ১০০টি কাজ
    new ThreadPoolExecutor.CallerRunsPolicy());   // সব ভরা হলে কী হবে

এখানে একটি ক্রম অনেককে অবাক করে। নতুন কাজ এলে: (১) core-এর চেয়ে কম thread চললে নতুন thread; (২) নইলে সারিতে; (৩) সারি ভরা হলে তবেই max পর্যন্ত বাড়তি thread; (৪) তাও ভরা হলে কাজটি প্রত্যাখ্যাত। অর্থাৎ সীমাহীন সারি দিলে max কখনো কাজেই লাগে না। প্রত্যাখ্যানের নিয়মকে বলে rejection policy:

• AbortPolicy (ডিফল্ট) — RejectedExecutionException ছোড়ে। বন্ধ করা pool-এ submit করলেও এটাই আসে।
• CallerRunsPolicy — যে thread জমা দিচ্ছিল, সে নিজেই কাজটা চালায়। ফলে জমা দেওয়ার গতি নিজে থেকেই কমে — চাপ সামলানোর সহজ উপায়।
• DiscardPolicy / DiscardOldestPolicy — নতুন বা সবচেয়ে পুরোনো কাজটি চুপচাপ ফেলে দেয়। শুধু তখনই, যখন কাজ হারালে ক্ষতি নেই (যেমন লাইভ স্কোরের পুরোনো আপডেট)।

৩ · Runnable বনাম Callable, আর Future — ফলের রসিদ

কাজ দুই ধরনের। Runnable-এর method void run() — কিছু ফেরত দেয় না, checked exception ছুড়তে পারে না। Callable<V>-এর method V call() throws Exception — একটি ফল দেয়, আর যেকোনো exception ছুড়তে পারে। দুটিই পর্ব ৩৪-এর functional interface, তাই lambda দিয়ে লেখা যায়। Lambda কিছু ফেরত দিলে submit তাকে Callable ধরে নেয়; কিছু ফেরত না দিলে সেটি Runnable, আর তখন ভেতরে checked exception ছোড়া যায় না।

Runnable-এর ভেতরে checked exception ছুড়লে compiler থামিয়ে দেয়:

class Main {
    public static void main(String[] args) {
        Runnable task = () -> {
            Thread.sleep(100);
        };
    }
}

Main.java:4: error: unreported exception InterruptedException; must be caught or declared to be thrown
            Thread.sleep(100);
                        ^
1 error

পর্ব ২৯-এর checked exception-এর নিয়ম: run() কিছু ঘোষণা করে না, তাই ভেতরে try-catch লাগবে। একই lambda Callable হলে (যেমন () -> { Thread.sleep(100); return 1; }) কোনো সমস্যা নেই।

Future — যে ফল পরে আসবে, তার রসিদ

দর্জির দোকানে জামা বানাতে দিলে দর্জি একটি রসিদ দেন। জামা তখনই হয় না, কিন্তু রসিদ হাতে আপনি অন্য কাজে যেতে পারেন। পরে রসিদ দেখিয়ে জামা নেবেন — তৈরি না হলে দোকানে বসে অপেক্ষা। pool.submit(callable) ঠিক তেমনি সাথে সাথে একটি Future<V> দেয়। এর method:

• get() — ফল না আসা পর্যন্ত বর্তমান thread আটকে থাকে (blocking), তারপর ফল দেয়।
• get(2, TimeUnit.SECONDS) — সর্বোচ্চ দুই সেকেন্ড অপেক্ষা; তার মধ্যে না এলে java.util.concurrent.TimeoutException (কোনো বার্তা ছাড়া)। কাজটি কিন্তু থামে না, চলতেই থাকে।
• isDone() — অপেক্ষা না করে জানায় শেষ হয়েছে কি না (সফল, ব্যর্থ বা বাতিল — তিনটিতেই true).
• cancel(true) — বাতিল; শুরু না হলে কখনো চলবে না, চলমান হলে thread-কে interrupt পাঠায়। এরপর get() ছোড়ে CancellationException.

Future<Integer> f = pool.submit(() -> countRows("jan.csv"));
submit ফিরে এল সাথে সাথে — কাজ এখনো শুরুই হয়নি
মূল thread পরের লাইনে চলে যেতে পারে

f.isDone()     → false
f.cancel(false) → true হতো, আর কাজটি কখনো চলত না
এখন f.get() ডাকলে মূল thread আটকে যাবে

Worker-2: call() → ফাইল পড়া চলছে …
f.isDone() → false
cancel(true) দিলে Worker-2-কে interrupt পাঠানো হতো

call() ফেরত দিল 120
Future-এর ভেতরে রাখা: 120
f.isDone() → true

int n = f.get();   → 120, কোনো অপেক্ষা ছাড়াই
কাজ শেষ হওয়ার আগে ডাকলে এই লাইনেই অপেক্ষা করত

call() ছুড়ল IllegalStateException("payment failed")
Future সেটা জমা রাখে — কেউ get() না ডাকলে কেউ জানবেও না!
f.get() → ExecutionException
  e.getCause() → IllegalStateException: payment failed

মোড়ানো exception না ধরলে stack trace দেখতে এমন — দুই অংশে:

import java.util.concurrent.*;

class Main {
    public static void main(String[] args) throws Exception {
        try (ExecutorService pool = Executors.newFixedThreadPool(2)) {
            Future<Integer> f = pool.submit(() -> 100 / 0);
            System.out.println(f.get());
        }
    }
}

Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zero
	at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
	at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191)
	at Main.main(Main.java:7)
Caused by: java.lang.ArithmeticException: / by zero
	at Main.lambda$main$0(Main.java:6)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)

প্রথম অংশ বলছে কোথায় ধরা পড়ল — মূল thread-এ, লাইন ৭-এর get()-এ। Caused by অংশ বলছে কোথায় ঘটেছিল — worker thread-এ, লাইন ৬-এর lambda-য়, ThreadPoolExecutor.runWorker-এর ভেতর দিয়ে। পর্ব ২৯-এর exception chaining-এর বাস্তব রূপ। Debug করার সময় সবসময় Caused by পর্যন্ত পড়ুন।

৪ · অনেক কাজ একসাথে, সঠিকভাবে বন্ধ, আর pool কত বড়

একটি রিপোর্ট জেনারেটর ভাবুন: চারটি মাসিক CSV ফাইল, প্রতিটি পড়তে সময় লাগে, আর শেষে মোট সারির সংখ্যা চাই। একটি ফাইল নেই — সেই ভুল যেন পুরো রিপোর্ট না থামায়। নিয়ম দুটি: আগে সব কাজ জমা দিন, তারপর ফল নিন; আর ফল নিন জমার ক্রমে, যাতে আউটপুট প্রতিবার একই থাকে।

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

class Main {
    static final Map<String, Integer> ROWS = Map.of("jan.csv", 120, "feb.csv", 95, "mar.csv", 140);

    static int countRows(String file) throws InterruptedException {
        Thread.sleep(100);                                   // ধীর ফাইল পড়ার ভান
        Integer rows = ROWS.get(file);
        if (rows == null) throw new IllegalStateException("ফাইল পাওয়া যায়নি: " + file);
        return rows;
    }
    public static void main(String[] args) throws InterruptedException {
        List<String> files = List.of("jan.csv", "feb.csv", "apr.csv", "mar.csv");
        int total = 0;
        try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
            List<Future<Integer>> receipts = new ArrayList<>();
            for (String f : files) receipts.add(pool.submit(() -> countRows(f)));  // সব আগে জমা
            for (int i = 0; i < files.size(); i++) {
                try {
                    int n = receipts.get(i).get();                // জমার ক্রমেই ফল নেওয়া
                    total += n;
                    System.out.println(files.get(i) + " -> " + n);
                } catch (ExecutionException e) {
                    System.out.println(files.get(i) + " -> ব্যর্থ: " + e.getCause().getMessage());
                }
            }
        }                                                         // close(): সব কাজ শেষের অপেক্ষা
        System.out.println("মোট সারি: " + total);
    }
}

invokeAll আর invokeAny

ওপরের “সব জমা, তারপর সব ফল” ধাঁচটি এত প্রচলিত যে এর জন্য তৈরি method আছে। invokeAll(tasks) সব কাজ জমা দেয়, সবগুলো শেষ হওয়া পর্যন্ত অপেক্ষা করে, আর Future-এর তালিকা দেয় — কাজের তালিকার ঠিক একই ক্রমে, কে আগে শেষ হলো তা নির্বিশেষে। invokeAny(tasks) সব জমা দেয়, যেটি প্রথম সফলভাবে শেষ হয় তার ফল দেয়, আর বাকিগুলো বাতিল করে। যেমন একই তথ্য তিনটি mirror সার্ভারে চাওয়া — যে আগে দেয়। (এর ফল কোন কাজের, তা সময়ের ওপর নির্ভর করে, তাই টেস্টে ছাপার আগে ভাবুন।)

বন্ধ করা — shutdown, shutdownNow, awaitTermination, close

Pool-এর worker-রা সাধারণ thread — daemon thread নয় — আর খালি থাকলে তারা নতুন কাজের অপেক্ষায় বসে থাকে। JVM তখনই শেষ হয়, যখন সব non-daemon thread শেষ — তাই pool বন্ধ না করলে main শেষ হলেও প্রোগ্রাম চলতেই থাকে, কিছু ছাপে না, শুধু ঝুলে থাকে। বন্ধ করার চারটি উপায়:

Java 19 থেকে ExecutorService পর্ব ২৯-এর AutoCloseable, তাই try-with-resources-এ রাখা যায় — ওপরের দুটি প্রোগ্রাম তা-ই করেছে। Java 21-এ এটাই সবচেয়ে নিরাপদ অভ্যাস। পুরোনো কোডে দেখবেন হাতে লেখা ধাঁচ: shutdown(), তারপর awaitTermination(30, SECONDS), সময়মতো শেষ না হলে shutdownNow(). একটি বাস্তব উদাহরণ — অর্ডার পাঠানোর পর হাজার ক্রেতাকে SMS:

try (ExecutorService smsPool = Executors.newFixedThreadPool(10)) {
    for (Customer c : customers)
        smsPool.execute(() -> smsGateway.send(c.phone(), "আপনার অর্ডার পাঠানো হয়েছে"));
}   // এখানে পৌঁছালে সব SMS-এর চেষ্টা শেষ

এখানে submit নয়, execute — কারণ ফল দরকার নেই। পার্থক্যটা জরুরি: submit-এর কাজ exception ছুড়লে সেটা Future-এ জমা থাকে, আর কেউ get() না ডাকলে কোথাও কিছু দেখা যায় না। execute-এর কাজের uncaught exception অন্তত stderr-এ stack trace হিসেবে আসে। আরও ভালো: কাজের ভেতরেই try-catch দিয়ে ব্যর্থ নম্বরগুলো লগ করুন, যাতে পরে আবার পাঠানো যায়। আর gateway যদি সেকেন্ডে ২০টির বেশি SMS নিতে না দেয়? Pool-এর আকারই তখন একটি সীমা — ১০টি thread মানে একসাথে সর্বোচ্চ ১০টি অনুরোধ।

Pool কত বড় হবে?

• CPU-নির্ভর কাজ (ছবি resize, হিসাব, কম্প্রেশন): thread সারাক্ষণ CPU ব্যবহার করে। Core-এর চেয়ে বেশি thread দিলে লাভ নেই, শুধু context switch. আকার ≈ core সংখ্যা: Runtime.getRuntime().availableProcessors().
• I/O-নির্ভর কাজ (ডেটাবেস, HTTP, ফাইল): thread বেশিরভাগ সময় উত্তরের অপেক্ষায় বসে থাকে, CPU খালি। তাই core-এর চেয়ে অনেক বেশি thread লাগে।

কত বেশি? Brian Goetz-এর Java Concurrency in Practice (২০০৬)-এর হিসাব: thread ≈ core × (১ + অপেক্ষার সময় ÷ হিসাবের সময়). ৮ core, প্রতিটি অনুরোধে ৯০ ms ডেটাবেসের অপেক্ষা আর ১০ ms হিসাব হলে ৮ × (১ + ৯) = ৮০. অন্য দিক থেকে পর্ব ৩২-এর Little-এর সূত্র L = λW: সেকেন্ডে ৪০০ অনুরোধ, প্রতিটি ০.১ সেকেন্ড থাকে, তাহলে গড়ে ৪০টি অনুরোধ একসাথে ভেতরে — প্রতিটি একটি thread আটকে রাখলে অন্তত ৪০টি thread লাগবে। দুটোই শুরুর অনুমান; আসল সংখ্যা ঠিক হয় load test করে। আর অপেক্ষা যখন খুব বেশি — হাজার হাজার একসাথে HTTP কল — তখন platform thread-এর pool নয়, পর্ব ৪২-এর virtual thread.

৫ · thread-দের সমন্বয় — CountDownLatch, Semaphore, CyclicBarrier, BlockingQueue

কখনো ফল নয়, শৃঙ্খলা দরকার: “তিনটি সার্ভিস চালু না হওয়া পর্যন্ত অপেক্ষা করো”, “একসাথে তিনটির বেশি ডাউনলোড নয়”। java.util.concurrent-এ এর জন্য তৈরি হাতিয়ার আছে, যাতে wait/notify হাতে লিখতে না হয়।

CountDownLatch — গোনা শূন্য হলে দরজা খোলে

CountDownLatch একটি সংখ্যা দিয়ে শুরু হয়। প্রতিটি কাজ শেষে countDown() সংখ্যাটি এক কমায়; await() সংখ্যা শূন্য না হওয়া পর্যন্ত অপেক্ষা করে। একবার শূন্য হলে আর ফেরানো যায় না — একবারের দরজা।

CountDownLatch ready = new CountDownLatch(3);
for (String service : List.of("database", "cache", "payment"))
    pool.execute(() -> { connect(service); ready.countDown(); });
ready.await();                       // তিনটিই চালু হলে তবেই সামনে
System.out.println("অ্যাপ প্রস্তুত");

বাস্তবে connect ব্যর্থ হলেও যেন গোনা কমে, তাই countDown() রাখুন finally-তে, আর await(10, TimeUnit.SECONDS) দিয়ে সময়সীমা দিন — নইলে একটি ব্যর্থতায় অ্যাপ চিরকাল আটকে থাকবে।

Semaphore — একসাথে সর্বোচ্চ কয়টি

একটি বাইরের API বলেছে: একসাথে তিনটির বেশি অনুরোধ পাঠালে আপনাকে আটকে দেবে। আপনার pool-এ thread অনেক, কারণ অন্য কাজও আছে। Semaphore এখানে টিকিট-ব্যবস্থা: শুরুতে ৩টি টিকিট (permit). acquire() একটি টিকিট নেয় — না থাকলে অপেক্ষা; release() টিকিট ফেরত দেয়।

1Semaphore gate = new Semaphore(3);2gate.acquire();          // টিকিট নাও, না থাকলে অপেক্ষা3try { download(url); }4finally { gate.release(); }   // টিকিট ফেরত

CyclicBarrier আর BlockingQueue — সংক্ষেপে

CyclicBarrier হলো মিলনস্থল: n-টি thread প্রত্যেকে await() ডাকে, আর সবাই না পৌঁছানো পর্যন্ত কেউ এগোয় না। তারপর বাধা আবার শুরুতে ফেরে — “cyclic”, বারবার ব্যবহার করা যায়। যেমন একটি simulation-এর প্রতিটি ধাপে চারটি thread মানচিত্রের চার অংশ হিসাব করে; সবাই শেষ করলে তবেই পরের ধাপ। আর পর্ব ৩২-এর BlockingQueue দিয়ে producer-consumer: একদল thread অর্ডার put করে, আরেকদল take করে প্রক্রিয়া করে; সারি ভরা হলে producer, খালি হলে consumer নিজে থেকেই অপেক্ষা করে। আসলে প্রতিটি ThreadPoolExecutor-এর ভেতরে ঠিক এই জিনিসটিই আছে — worker-রা হলো consumer, আর আপনার submit হলো producer.

৬ · CompletableFuture — “এটা শেষ হলে ওটা করো”

সাধারণ Future-এর বড় দুর্বলতা: ফল দিয়ে কিছু করতে হলে কাউকে get() ডেকে অপেক্ষা করতেই হয়। “দাম এলে ছাড় হিসাব করো, তারপর পেজ বানাও” — এই বাক্যটি Future দিয়ে লেখা যায় না, অপেক্ষা ছাড়া। Java 8-এর CompletableFuture ঠিক এটাই দেয়: আপনি আগে থেকেই শিকল বর্ণনা করেন, আর প্রতিটি ধাপ আগেরটি শেষ হলে নিজে থেকেই চলে। কোনো thread বসে অপেক্ষা করে না।

CompletableFuture<Void> done = CompletableFuture
    .supplyAsync(() -> priceService.get(productId), io)   // ব্যাকগ্রাউন্ডে শুরু
    .thenApply(price -> price * 90 / 100)                  // ফল এলে: ১০% ছাড়
    .thenAccept(p -> System.out.println("দাম: " + p));     // শেষ ধাপ: ব্যবহার

• supplyAsync(supplier, executor) — একটি Supplier ব্যাকগ্রাউন্ডে চালায়, সাথে সাথে CompletableFuture দেয়। ফল না লাগলে runAsync(runnable).
• thenApply(fn) — ফলকে রূপান্তর; পর্ব ৩৫-এর stream-এর map-এর মতো।
• thenAccept(consumer) — ফল ব্যবহার করে, কিছু ফেরত দেয় না (CompletableFuture<Void>).
• join() — শেষ পর্যন্ত অপেক্ষা করে ফল দেয়; get()-এর মতোই, কিন্তু checked exception ছোড়ে না — ভুল আসে unchecked CompletionException-এ মোড়ানো। get() ছোড়ে checked ExecutionException. শিকলের একদম শেষে, সাধারণত একবারই।

thenCompose বনাম thenCombine

দুটি সার্ভিস কল কীভাবে সম্পর্কিত, তার ওপর method নির্ভর করে:

var price = supplyAsync(() -> priceService(), io);
var stock = supplyAsync(() -> stockService(), io);
দুটি লাইনই সাথে সাথে ফিরে এল — দুটি worker কাজ শুরু করল

var page = price.thenCombine(stock, (p, s) -> …)
                .thenApply(Main::format);
এটা শুধু পরিকল্পনা — কোনো ফল আসেনি, কোনো thread অপেক্ষা করছে না

১০০ ms: stock = 7     (একা এসে অপেক্ষা)
২০০ ms: price = 1250
দুটো আসামাত্র thenCombine চলল → (1250, 7)

thenApply(format) → "দাম 1250 টাকা, স্টকে 7টি"
যে thread আগের ধাপ শেষ করেছে, সাধারণত সে-ই এটা চালায়

String html = page.join();
মোট সময় ≈ ২০০ ms (ধীরতম কল), ৩০০ ms নয়

অনেকগুলো একসাথে — allOf আর anyOf

CompletableFuture.allOf(a, b, c) একটি CompletableFuture<Void> দেয়, যা সবগুলো শেষ হলে শেষ হয়। এর নিজের কোনো ফল নেই (join() দেয় null); ফল নিতে হয় প্রতিটি থেকে আলাদাভাবে, ক্রমমতো: allOf(a, b, c).thenApply(v -> List.of(a.join(), b.join(), c.join())) — এখানে join() আর অপেক্ষা করে না, কারণ সব শেষ। anyOf(...) দেয় যেটি প্রথম শেষ হয় তার ফল, টাইপ Object.

৭ · ভুল, timeout আর fan-out/fan-in — একটি পণ্যের পেজ

একটি ই-কমার্স পণ্যের পেজে চারটি তথ্য লাগে, চারটি আলাদা সার্ভিস থেকে: দাম, স্টক, ডেলিভারির সময়, রেটিং। একটির পর একটি ডাকলে সময় যোগ হয়। একসাথে ছড়িয়ে দেওয়া (fan-out), তারপর এক জায়গায় জোড়া (fan-in) — এই fan-out/fan-in ধাঁচে সময় হয় ধীরতম কলের সমান। কিন্তু বাস্তবের দুটি ঝামেলা: কোনো সার্ভিস খুব ধীর, কোনোটি ভেঙে পড়েছে। পেজ তবুও দেখাতে হবে।

• exceptionally(ex -> বিকল্প) — আগের ধাপ ব্যর্থ হলে একটি বিকল্প মান দেয়; সফল হলে কিছু করে না।
• handle((value, ex) -> …) — সফল বা ব্যর্থ, দুই ক্ষেত্রেই চলে; একটি null থাকে। ফল আর ভুল দুটো দেখে সিদ্ধান্ত নিতে।
• orTimeout(t, unit) (Java 9) — সময়ের মধ্যে না এলে TimeoutException দিয়ে ব্যর্থ করে। completeOnTimeout(value, t, unit) (Java 9) — সময়ের মধ্যে না এলে বিকল্প মান দিয়ে সফল করে।

import java.util.concurrent.*;

class Main {
    static <T> T slowCall(int ms, T value) {                 // ধীর নেটওয়ার্ক কলের ভান
        try { Thread.sleep(ms); } catch (InterruptedException e) { throw new RuntimeException(e); }
        return value;
    }
    static int ratingService() { slowCall(50, 0); throw new IllegalStateException("রেটিং সার্ভার বন্ধ"); }

    public static void main(String[] args) {
        try (ExecutorService io = Executors.newFixedThreadPool(4)) {
            var price = CompletableFuture.supplyAsync(() -> slowCall(200, 1250), io);
            var stock = CompletableFuture.supplyAsync(() -> slowCall(100, 7), io);
            var delivery = CompletableFuture.supplyAsync(() -> slowCall(900, "কাল"), io)
                .completeOnTimeout("৩-৫ দিন (আনুমানিক)", 300, TimeUnit.MILLISECONDS);
            var rating = CompletableFuture.supplyAsync(Main::ratingService, io)
                .thenApply(r -> r + "/5")
                .exceptionally(ex -> "রেটিং নেই (" + ex.getClass().getSimpleName() + ")");

            String page = price.thenCombine(stock, (p, s) -> "দাম " + p + " টাকা, স্টকে " + s + "টি")
                .thenCombine(delivery, (ps, d) -> ps + "\nডেলিভারি: " + d)
                .thenCombine(rating, (pd, r) -> pd + "\n" + r)
                .join();
            System.out.println(page);
        }
    }
}

টাকার লেনদেনে timeout — সাবধান

পেমেন্ট গেটওয়ের কলে তিন সেকেন্ডের সীমা দেওয়া স্বাভাবিক:

CompletableFuture<String> status = CompletableFuture
    .supplyAsync(() -> gateway.charge(orderId, amount), paymentPool)
    .orTimeout(3, TimeUnit.SECONDS)
    .exceptionally(ex -> "PENDING");        // গ্রাহককে “যাচাই চলছে” দেখান

কিন্তু ওপরের উদাহরণের শিক্ষা মনে রাখুন: timeout শুধু আপনার অপেক্ষা থামায়, গেটওয়ের কাজ নয়। টাকা হয়তো ঠিকই কেটেছে, উত্তর আসতে দেরি হয়েছে। তাই ফলব্যাক কখনো “ব্যর্থ, আবার চেষ্টা করুন” নয় — তাতে দুবার টাকা কাটতে পারে। ঠিক উপায় হলো “PENDING” অবস্থা রেখে পরে গেটওয়ের কাছে লেনদেনের অবস্থা যাচাই করা, আর প্রতিটি অনুরোধে একটি অনন্য idempotency key পাঠানো, যাতে একই অনুরোধ দুবার এলে গেটওয়ে একবারই কাটে। bKash বা কার্ড পেমেন্টে “টাকা কেটেছে কিন্তু অর্ডার হয়নি” অভিযোগের বড় একটি উৎস ঠিক এই জায়গা।

৮ · নির্দিষ্ট সময় পরপর কাজ, structured concurrency, আর অন্য ভাষার সাথে তুলনা

ScheduledExecutorService

প্রতি ৩০ সেকেন্ডে অপেক্ষমাণ পেমেন্টগুলোর অবস্থা দেখা, প্রতি রাতে একটি রিপোর্ট — এর জন্য ScheduledExecutorService:

ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);
timer.schedule(() -> sendReminder(cart), 10, TimeUnit.MINUTES);          // একবার, ১০ মিনিট পরে
timer.scheduleAtFixedRate(this::checkPendingPayments, 0, 30, TimeUnit.SECONDS);
// অ্যাপ বন্ধের সময়: timer.shutdown();

scheduleAtFixedRate শুরুর সময় ধরে গোনে (০, ৩০, ৬০ সেকেন্ড…); কাজ দেরি করলে পরেরটা দেরিতে শুরু হয়, কিন্তু একই কাজ কখনো একসাথে দুবার চলে না। scheduleWithFixedDelay গোনে আগের কাজ শেষ হওয়ার পর থেকে। একটি ফাঁদ: পর্যায়ক্রমিক কাজের কোনো একবার exception ছুড়লে পরের সব বার নীরবে বন্ধ হয়ে যায় — কোনো লগ ছাড়া। তাই কাজের পুরো শরীর try-catch-এ মুড়ে ভুল লগ করুন। পুরোনো java.util.Timer একটি thread-এ চলে আর একটি exception-এ পুরোটাই থেমে যায়; নতুন কোডে এটি এড়িয়ে চলুন।

Structured concurrency — একটি প্রথম ঝলক

CompletableFuture-এর একটি সমস্যা ওপরে দেখেছেন: একটি কাজ ব্যর্থ হলে বা timeout হলে অন্য কাজগুলো চলতেই থাকে, কেউ তাদের বাতিল করে না। Structured concurrency-র ধারণা: একটি কাজ থেকে যতগুলো উপ-কাজ জন্ম নেয়, সবগুলো সেই কাজের একটি code block-এর ভেতরে শুরু ও শেষ হবে — block থেকে বেরোনোর আগে সব শেষ বা বাতিল। Java 21-এ এটি preview feature (JEP 453), চালাতে --enable-preview লাগে, আর পরের সংস্করণগুলোতে API বদলেছে। Java 21-এর রূপ:

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {   // Java 21, preview
    Subtask<Integer> price = scope.fork(() -> priceService(id));
    Subtask<Integer> stock = scope.fork(() -> stockService(id));
    scope.join().throwIfFailed();      // একটি ব্যর্থ হলে বাকিগুলো বাতিল
    return render(price.get(), stock.get());
}

এটি virtual thread-এর সাথে মিলিয়ে নকশা করা, তাই বিস্তারিত পর্ব ৪২-এ। এখানে শুধু ধারণাটা মনে রাখুন: একসাথে চলা কাজের জীবনকাল code block-এর কাঠামোর সাথে বাঁধা থাকলে কোনো কাজ “হারিয়ে” পেছনে চলতে থাকে না।

৯ · AI-এর যুগে — ঝুলে থাকা প্রোগ্রাম আর গিলে ফেলা exception

Concurrency-র কোড AI খুব আত্মবিশ্বাসের সাথে লেখে, আর সেটা প্রায়ই ছোট টেস্টে ঠিকঠাক চলে। ভুল ধরা পড়ে ভিড়ের সময়, সার্ভারে, রাত দুটোয়। AI-এর লেখা executor কোডে বারবার যা দেখা যায়:

AI-এর concurrency কোড হাতে পেলে — যাচাইয়ের চেকলিস্ট

• প্রতিটি executor-এর আকার সীমিত, আর কোথায় বন্ধ হচ্ছে তা কোডে স্পষ্ট
• কোনো লুপে submit(...).get() একসাথে নেই
• প্রতিটি submit-এর Future কেউ না কেউ get করছে, অথবা কাজের ভেতরেই ভুল লগ হচ্ছে
• supplyAsync/runAsync-এ I/O থাকলে নিজের executor দেওয়া আছে
• প্রতিটি অপেক্ষার (get, join, await, acquire) সময়সীমা বা স্পষ্ট কারণ আছে
• Semaphore.release() আর countDown() finally-তে
• timeout-এর ফলব্যাক টাকা বা ডেটার ক্ষেত্রে নিরাপদ — আসল কাজ যে তখনও চলছে, তা মাথায় রাখা হয়েছে