CompletableFuture-এর মৌলিক বিষয়: একটি অ্যাসিঙ্ক পাইপলাইন তৈরি করা

একটি Future যার সাথে কাজ যুক্ত করা যায়

CompletableFuture ইন্টারফেসটি Future ইমপ্লিমেন্ট করে, তাই আগের লেখার সবকিছু — get(), isDone(), cancel() — এখানেও একইভাবে কাজ করে। এটি যে অতিরিক্ত ক্ষমতা যোগ করে তা হলো: ফলাফল প্রস্তুত হলে কী ঘটবে তা নিবন্ধন (register) করার সুবিধা। কলিং থ্রেডকে বসে বসে ফলাফলের জন্য অপেক্ষা করতে বাধ্য না করেই এটি সম্ভব।

একটি পাইপলাইন শুরু করা: supplyAsync

CompletableFuture.supplyAsync(Supplier) একটি অ্যাসিঙ্ক্রোনাস কম্পিউটেশন শুরু করে। এটি ডিফল্টরূপে JVM-ব্যাপী সাধারণ থ্রেড পুল (common pool) ব্যবহার করে এবং তাৎক্ষণিকভাবে ভবিষ্যতের ফলাফলের একটি হ্যান্ডেল ফেরত দেয়।

CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> lookupUser(id));

উপরের কোডে কোনো থ্রেড ব্লক হয়নি। userFuture অবজেক্টটি তাৎক্ষণিকভাবে তৈরি হয়ে যায়; lookupUser কলটি ব্যাকগ্রাউন্ডে চলতে থাকে।

ফলাফল রূপান্তর করা: thenApply, thenAccept, thenRun

তিনটি কলব্যাক মেথড বেশিরভাগ চাহিদা পূরণ করে। এদের পার্থক্য কেবল তারা কী গ্রহণ করে ও কী ফেরত দেয় তার ওপর ভিত্তি করে।

• thenApply: ফলাফল গ্রহণ করে এবং একটি নতুন মান তৈরি করে।
• thenAccept: ফলাফল গ্রহণ করে কিন্তু কিছু ফেরত দেয় না (পার্শ্ব-প্রতিক্রিয়ার জন্য)।
• thenRun: ফলাফলকে সম্পূর্ণ উপেক্ষা করে এবং আগের ধাপ শেষ হলেই কেবল চালানো হয়।

CompletableFuture<String> nameFuture = userFuture.thenApply(User::getName);
nameFuture.thenAccept(name -> System.out.println("Loaded " + name));
nameFuture.thenRun(() -> System.out.println("Loading finished"));

প্রতিটি কল একটি নতুন CompletableFuture রিটার্ন করে, যা চেইনিং সম্ভব করে তোলে: একটি ধাপের রিটার্ন ভ্যালুই পরবর্তী ধাপের রিসিভার হয়ে যায়।

একাধিক ধাপ চেইন করা

যেহেতু প্রতিটি then* মেথড আরেকটি CompletableFuture ফেরত দেয়, একটি সম্পূর্ণ পাইপলাইনকে একটিমাত্র ফ্লুয়েন্ট এক্সপ্রেশন হিসেবে লেখা যায়। মাঝখানে কোনো get() কলের প্রয়োজন পড়ে না।

CompletableFuture<String> greeting = CompletableFuture
    .supplyAsync(() -> lookupUser(id))
    .thenApply(User::getName)
    .thenApply(name -> "Hello, " + name);

আগের লেখায় হাতে তৈরি Future-এর সংস্করণের সাথে এটি তুলনা করুন: সেখানে প্রথম ধাপের পর দ্বিতীয় ধাপ যোগ করতে get() দিয়ে ব্লক করে আবার ম্যানুয়ালি আরেকটি টাস্ক সাবমিট করতে হতো। এখানে চেইনিং API-এর মধ্যেই তৈরি, এবং কোনো থ্রেডই ধাপগুলোর মাঝে ব্লক হতে বাধ্য হয় না।

কলব্যাকগুলো কাউকে ব্লক না করেই চলে

একটি সূক্ষ্ম কিন্তু গুরুত্বপূর্ণ বিষয়: thenApply নিবন্ধন করামাত্র কলব্যাকটি চলে না, যদি না Future ইতিমধ্যে সম্পন্ন হয়ে থাকে। যদি ব্যাকগ্রাউন্ডের কাজ শেষ না হয়ে থাকে, কলব্যাকটি সংরক্ষিত থাকে এবং পরে সেই থ্রেডের মাধ্যমে চালিত হয় যে Future-টি সম্পন্ন করে — যা প্রায়শই কমন পুলের কোনো ওয়ার্কার থ্রেড, thenApply কল করা থ্রেড নয়।

যদি আপনি কলব্যাক যুক্ত করার সময় Future আগেই সম্পন্ন হয়ে থাকে, তাহলে সেটি কলিং থ্রেডেই তাৎক্ষণিকভাবে চলতে পারে। উভয় ক্ষেত্রেই, যে থ্রেড পাইপলাইনটি তৈরি করেছিল তাকে কখনো অপেক্ষায় ব্লক হতে হয়নি।

কোডটি আসলে কোথায় চলে?

ডিফল্টভাবে, supplyAsync এবং সাধারণ then* মেথডগুলো কমন ForkJoinPool-এ চলে। এটি JVM-ব্যাপী একটি শেয়ার্ড পুল, যার আকার CPU কোরের সংখ্যার সমান। CPU-নির্ভর রূপান্তরের জন্য এটি একটি যুক্তিসঙ্গত ডিফল্ট।