Concurrency Utilities (java.util.concurrent)

java.util.concurrent প্যাকেজটি জাভার কনকারেন্সি ফ্রেমওয়ার্কের একটি গুরুত্বপূর্ণ অংশ। এটি মাল্টিথ্রেডেড প্রোগ্রামিংকে সহজ এবং কার্যকর করতে বিভিন্ন ক্লাস এবং ইন্টারফেস সরবরাহ করে। এই প্যাকেজে থাকা ইউটিলিটিগুলি সঠিকভাবে ব্যবহার করলে ডেটা সিঙ্ক্রোনাইজেশন, থ্রেড ম্যানেজমেন্ট, এবং পারফরম্যান্স উন্নত করা যায়।

১. Executor Framework

Executor Framework থ্রেড ম্যানেজমেন্টের জন্য ব্যবহার করা হয়। এটি সরাসরি Thread ক্লাস ব্যবহার না করে একটি থ্রেড পুল তৈরি করে এবং কাজ পরিচালনা করে।

উদাহরণ: ExecutorService

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(3); // ৩ থ্রেডের পুল

        Runnable task = () -> {
            System.out.println("Task executed by: " + Thread.currentThread().getName());
        };

        // থ্রেড পুলে কাজ জমা দেওয়া
        for (int i = 0; i < 5; i++) {
            executor.submit(task);
        }

        executor.shutdown(); // থ্রেড পুল বন্ধ করা
    }
}

২. Callable এবং Future

Callable ইন্টারফেস থ্রেড থেকে একটি ফলাফল রিটার্ন করার জন্য ব্যবহৃত হয়। Future ইন্টারফেস এই ফলাফল পেতে ব্যবহৃত হয়।

উদাহরণ: Callable এবং Future

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class CallableExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        Callable<String> task = () -> {
            Thread.sleep(2000);
            return "Task Completed!";
        };

        Future<String> future = executor.submit(task);

        try {
            System.out.println("Result: " + future.get()); // ফলাফল পেতে
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
        }
    }
}

৩. Thread-safe Collections

java.util.concurrent প্যাকেজ থ্রেড-সেফ কালেকশন সরবরাহ করে, যেমন:

• ConcurrentHashMap
• CopyOnWriteArrayList
• LinkedBlockingQueue

উদাহরণ: ConcurrentHashMap

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapExample {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        map.put("One", 1);
        map.put("Two", 2);

        map.forEach((key, value) -> System.out.println(key + ": " + value));
    }
}

৪. BlockingQueue

BlockingQueue একটি থ্রেড-সেফ কিউ, যা প্রযোজক-গ্রাহক প্যাটার্নে ব্যবহৃত হয়।

উদাহরণ: ArrayBlockingQueue

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        Runnable producer = () -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    queue.put(i);
                    System.out.println("Produced: " + i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        Runnable consumer = () -> {
            try {
                while (true) {
                    Integer value = queue.take();
                    System.out.println("Consumed: " + value);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        new Thread(producer).start();
        new Thread(consumer).start();
    }
}

৫. Locks (ReentrantLock)

ReentrantLock সিঙ্ক্রোনাইজেশনের জন্য একটি উন্নত বিকল্প। এটি থ্রেডে আরও বেশি নিয়ন্ত্রণ প্রদান করে।

উদাহরণ: ReentrantLock

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {
    private final Lock lock = new ReentrantLock();

    public void sharedMethod() {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + " is executing...");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ReentrantLockExample example = new ReentrantLockExample();

        Runnable task = example::sharedMethod;

        new Thread(task, "Thread-1").start();
        new Thread(task, "Thread-2").start();
    }
}

৬. Phaser

Phaser হলো একটি উন্নত CountDownLatch, যা ডাইনামিক সংখ্যা পরিবর্তন করতে পারে।

উদাহরণ: Phaser

import java.util.concurrent.Phaser;

public class PhaserExample {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(1); // মেইন থ্রেড রেজিস্টার
        int tasks = 3;

        for (int i = 0; i < tasks; i++) {
            int taskNumber = i + 1;
            phaser.register(); // প্রতিটি কাজ রেজিস্টার করা

            new Thread(() -> {
                System.out.println("Task " + taskNumber + " started.");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Task " + taskNumber + " completed.");
                phaser.arriveAndDeregister(); // কাজ শেষ
            }).start();
        }

        phaser.arriveAndAwaitAdvance(); // অপেক্ষা করা সব কাজ শেষ হওয়ার জন্য
        System.out.println("All tasks completed.");
    }
}

৭. CountDownLatch

CountDownLatch নির্দিষ্ট সংখ্যক কাজ সম্পন্ন হওয়া পর্যন্ত অপেক্ষা করে।

উদাহরণ: CountDownLatch

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);

        Runnable task = () -> {
            try {
                System.out.println(Thread.currentThread().getName() + " is working.");
                Thread.sleep(1000);
                latch.countDown();
                System.out.println(Thread.currentThread().getName() + " finished.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        new Thread(task, "Thread-1").start();
        new Thread(task, "Thread-2").start();
        new Thread(task, "Thread-3").start();

        latch.await(); // সব কাজ শেষ না হওয়া পর্যন্ত অপেক্ষা
        System.out.println("All threads finished.");
    }
}

java.util.concurrent প্যাকেজ জাভাতে মাল্টিথ্রেডিং এবং কনকারেন্সি ব্যবস্থাপনাকে সহজ করে। এটি বিভিন্ন সমস্যা সমাধানে ব্যবহার করা হয়, যেমন:

• থ্রেড ম্যানেজমেন্ট: Executor Framework
• থ্রেড সিঙ্ক্রোনাইজেশন: Locks, CountDownLatch, Phaser
• থ্রেড-সেফ কালেকশন: ConcurrentHashMap, BlockingQueue

এই ইউটিলিটিগুলি সঠিকভাবে ব্যবহার করে কনকারেন্সি সহজ, কার্যকর, এবং নিরাপদ করা সম্ভব।

Java এর java.util.concurrent প্যাকেজ মাল্টিথ্রেডিং এবং কনকারেন্ট প্রোগ্রামিংয়ের জন্য বিভিন্ন শক্তিশালী টুলস এবং ইউটিলিটি সরবরাহ করে। এই প্যাকেজটি Java 5-এ পরিচিত হয়েছিল এবং এর পরে আরও উন্নত হয়েছে।

Utility Overview

java.util.concurrent প্যাকেজ প্রধানত নিম্নলিখিত বিষয়গুলো কভার করে:

1. Executors এবং Thread Pools
   • থ্রেড তৈরি এবং ম্যানেজমেন্ট সহজ করে।
2. Concurrency Utilities
   • Synchronization, Atomic Operations, এবং Locks।
3. Concurrent Collections
   • থ্রেড-সেফ ডেটা স্ট্রাকচার।
4. Blocking Queues
   • প্রোডিউসার-কনসিউমার প্যাটার্নে ব্যবহৃত হয়।
5. Synchronizers
   • Thread Coordination এবং Synchronization।

1. Executors এবং Thread Pools

Executors:

Executors API দিয়ে থ্রেড তৈরি এবং ম্যানেজ করা সহজ।

উদাহরণ:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorsExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(3);

        Runnable task = () -> {
            System.out.println(Thread.currentThread().getName() + " is executing task.");
        };

        for (int i = 0; i < 5; i++) {
            executor.submit(task);
        }

        executor.shutdown();
    }
}

Thread Pool Types:

• newFixedThreadPool(int nThreads): নির্দিষ্ট সংখ্যক থ্রেড।
• newCachedThreadPool(): প্রয়োজন অনুযায়ী থ্রেড তৈরি।
• newSingleThreadExecutor(): একক থ্রেড।

2. Concurrency Utilities

Atomic Classes:

Atomic অপারেশন করার জন্য ব্যবহার হয়।

উদাহরণ:

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicExample {
    private static AtomicInteger counter = new AtomicInteger(0);

    public static void main(String[] args) {
        Runnable incrementTask = () -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("Incremented: " + counter.incrementAndGet());
            }
        };

        Thread t1 = new Thread(incrementTask);
        Thread t2 = new Thread(incrementTask);

        t1.start();
        t2.start();
    }
}

ReentrantLock:

মাল্টিথ্রেডেড প্রোগ্রামে লকিং।

উদাহরণ:

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {
    private static ReentrantLock lock = new ReentrantLock();

    public static void main(String[] args) {
        Runnable task = () -> {
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName() + " is in critical section.");
            } finally {
                lock.unlock();
            }
        };

        Thread t1 = new Thread(task);
        Thread t2 = new Thread(task);

        t1.start();
        t2.start();
    }
}

3. Concurrent Collections

ConcurrentHashMap:

থ্রেড-সেফ হ্যাশম্যাপ।

উদাহরণ:

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapExample {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
        map.put("A", 1);
        map.put("B", 2);

        map.putIfAbsent("C", 3);
        map.computeIfPresent("A", (key, val) -> val + 10);

        System.out.println(map);
    }
}

CopyOnWriteArrayList:

থ্রেড-সেফ ArrayList।

উদাহরণ:

import java.util.concurrent.CopyOnWriteArrayList;

public class CopyOnWriteArrayListExample {
    public static void main(String[] args) {
        CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
        list.add("A");
        list.add("B");

        for (String item : list) {
            System.out.println(item);
        }
    }
}

4. Blocking Queues

ArrayBlockingQueue:

ফিক্সড সাইজের ব্লকিং কিউ।

LinkedBlockingQueue:

ডাইনামিক সাইজ ব্লকিং কিউ।

উদাহরণ:

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();

        Runnable producer = () -> {
            try {
                queue.put("Item1");
                System.out.println("Produced: Item1");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        Runnable consumer = () -> {
            try {
                System.out.println("Consumed: " + queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        new Thread(producer).start();
        new Thread(consumer).start();
    }
}

5. Synchronizers

CountDownLatch:

নির্দিষ্ট সংখ্যক থ্রেড শেষ হওয়ার অপেক্ষা।

উদাহরণ:

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);

        Runnable task = () -> {
            System.out.println(Thread.currentThread().getName() + " finished task.");
            latch.countDown();
        };

        new Thread(task).start();
        new Thread(task).start();
        new Thread(task).start();

        latch.await();
        System.out.println("All tasks finished.");
    }
}

CyclicBarrier:

একসঙ্গে একাধিক থ্রেডের কাজ শুরু বা শেষ।

উদাহরণ:

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("All parties reached the barrier."));

        Runnable task = () -> {
            try {
                System.out.println(Thread.currentThread().getName() + " is waiting at the barrier.");
                barrier.await();
                System.out.println(Thread.currentThread().getName() + " crossed the barrier.");
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        new Thread(task).start();
        new Thread(task).start();
        new Thread(task).start();
    }
}

Semaphore:

একসঙ্গে নির্দিষ্ট সংখ্যক থ্রেড রিসোর্সে অ্যাক্সেস।

java.util.concurrent প্যাকেজ মাল্টিথ্রেডেড প্রোগ্রামিংয়ের জন্য অত্যন্ত কার্যকরী। এটি থ্রেড ম্যানেজমেন্ট থেকে শুরু করে রিসোর্স শেয়ারিং, সিঙ্ক্রোনাইজেশন এবং ডেটা স্ট্রাকচারের নিরাপদ ব্যবস্থাপনা পর্যন্ত সবকিছু সহজ করে তোলে। সঠিক টুলসের ব্যবহার কনকারেন্সি সমস্যাগুলি সমাধান করতে সাহায্য করে এবং অ্যাপ্লিকেশনের কর্মক্ষমতা বাড়ায়।

জাভার কনকারেন্সি টুলকিটে এই ক্লাসগুলো ব্যবহারের মাধ্যমে মাল্টিথ্রেডিং প্রোগ্রাম সহজে পরিচালনা করা যায়। এখানে Executors, CountDownLatch, CyclicBarrier, এবং Semaphore এর ব্যবহার বিস্তারিতভাবে ব্যাখ্যা করা হলো।

১. Executors

Executors হলো একটি থ্রেড পুল তৈরি ও পরিচালনার জন্য জাভার একটি ক্লাস। এটি একাধিক থ্রেড তৈরি ও পরিচালনা করার কাজ সহজ করে।

ব্যবহার:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorsExample {
    public static void main(String[] args) {
        // একটি Fixed Thread Pool তৈরি
        ExecutorService executor = Executors.newFixedThreadPool(3);

        // কাজ জমা দিন
        for (int i = 1; i <= 5; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " is running in thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000); // কাজ করছে
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // থ্রেড পুল বন্ধ করুন
        executor.shutdown();
    }
}

আউটপুট (উদাহরণ):

Task 1 is running in thread pool-1-thread-1
Task 2 is running in thread pool-1-thread-2
Task 3 is running in thread pool-1-thread-3
Task 4 is running in thread pool-1-thread-1
Task 5 is running in thread pool-1-thread-2

২. CountDownLatch

CountDownLatch একাধিক থ্রেডকে সমন্বয় করতে ব্যবহার করা হয়। এটি থ্রেডগুলোকে নির্দিষ্ট সংখ্যক অপারেশন শেষ না হওয়া পর্যন্ত অপেক্ষা করায়।

ব্যবহার:

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3); // ৩টি কাজ শেষ হওয়া পর্যন্ত অপেক্ষা করবে

        for (int i = 1; i <= 3; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + " completed its task.");
                latch.countDown(); // কাউন্টডাউন
            }).start();
        }

        latch.await(); // মেইন থ্রেড অপেক্ষা করবে
        System.out.println("All tasks are completed. Main thread is proceeding.");
    }
}

আউটপুট:

Thread-1 completed its task.
Thread-2 completed its task.
Thread-3 completed its task.
All tasks are completed. Main thread is proceeding.

৩. CyclicBarrier

CyclicBarrier হলো একটি টুল, যা একাধিক থ্রেডকে নির্দিষ্ট পয়েন্টে অপেক্ষা করায় যতক্ষণ না সব থ্রেড পৌঁছে। এটি পুনরায় ব্যবহারযোগ্য।

ব্যবহার:

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    public static void main(String[] args) {
        int numberOfThreads = 3;
        CyclicBarrier barrier = new CyclicBarrier(numberOfThreads, () -> {
            System.out.println("All threads reached the barrier. Proceeding...");
        });

        for (int i = 1; i <= numberOfThreads; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + " reached the barrier.");
                try {
                    barrier.await(); // ব্যারিয়ারে অপেক্ষা
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

আউটপুট:

Thread-1 reached the barrier.
Thread-2 reached the barrier.
Thread-3 reached the barrier.
All threads reached the barrier. Proceeding...

৪. Semaphore

Semaphore হলো একটি টুল, যা নির্দিষ্ট সংখ্যক থ্রেডকে একবারে একটি রিসোর্স অ্যাক্সেস করার অনুমতি দেয়।

ব্যবহার:

import java.util.concurrent.Semaphore;

public class SemaphoreExample {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(2); // ২টি থ্রেড একবারে রিসোর্স অ্যাক্সেস করতে পারবে

        for (int i = 1; i <= 5; i++) {
            final int taskId = i;
            new Thread(() -> {
                try {
                    System.out.println("Task " + taskId + " is waiting for a permit.");
                    semaphore.acquire(); // পারমিট নেওয়া
                    System.out.println("Task " + taskId + " got a permit and is working.");
                    Thread.sleep(2000); // কাজ করছে
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("Task " + taskId + " released the permit.");
                    semaphore.release(); // পারমিট ছাড়া
                }
            }).start();
        }
    }
}

আউটপুট:

Task 1 is waiting for a permit.
Task 2 is waiting for a permit.
Task 1 got a permit and is working.
Task 2 got a permit and is working.
Task 3 is waiting for a permit.
Task 4 is waiting for a permit.
Task 5 is waiting for a permit.
Task 1 released the permit.
Task 3 got a permit and is working.
Task 2 released the permit.
Task 4 got a permit and is working.

একত্রে ব্যবহারের উদাহরণ

import java.util.concurrent.*;

public class ConcurrencyExample {
    public static void main(String[] args) throws InterruptedException {
        int numberOfThreads = 3;

        ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
        CountDownLatch latch = new CountDownLatch(numberOfThreads);
        CyclicBarrier barrier = new CyclicBarrier(numberOfThreads);
        Semaphore semaphore = new Semaphore(2);

        for (int i = 1; i <= numberOfThreads; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    semaphore.acquire();
                    System.out.println("Task " + taskId + " acquired semaphore.");
                    barrier.await();
                    System.out.println("Task " + taskId + " crossed the barrier.");
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();
                    latch.countDown();
                }
            });
        }

        latch.await();
        executor.shutdown();
        System.out.println("All tasks completed. Main thread proceeding.");
    }
}

আউটপুট:

Task 1 acquired semaphore.
Task 2 acquired semaphore.
Task 3 acquired semaphore.
Task 1 crossed the barrier.
Task 2 crossed the barrier.
Task 3 crossed the barrier.
All tasks completed. Main thread proceeding.

1. Executors: থ্রেড ম্যানেজমেন্ট সহজ করে।
2. CountDownLatch: থ্রেডগুলোর কাজ শেষ হওয়া পর্যন্ত মেইন থ্রেড অপেক্ষা করে।
3. CyclicBarrier: থ্রেডগুলোকে একত্রে কাজ শুরু করার জন্য সিঙ্ক্রোনাইজ করে।
4. Semaphore: নির্দিষ্ট সংখ্যক থ্রেডকে একবারে রিসোর্স অ্যাক্সেস করতে দেয়।

এই ক্লাসগুলো একত্রে ব্যবহার করে জটিল মাল্টিথ্রেডেড অ্যাপ্লিকেশন সহজে তৈরি করা যায়।

DelayQueue এবং PriorityBlockingQueue হল জাভার কনকারেন্সি ফ্রেমওয়ার্কে ব্যবহৃত বিশেষধরনের BlockingQueue। এদের ভূমিকা এবং ব্যবহারের উপায় নিচে বিস্তারিতভাবে ব্যাখ্যা করা হলো।

DelayQueue

DelayQueue হলো একটি ব্লকিং কিউ, যা এমন ডেটা সংরক্ষণ করে যেগুলো নির্দিষ্ট সময় পর অ্যাক্সেসযোগ্য হয়। এটি Delayed ইন্টারফেস ব্যবহার করে কাজ করে।

বৈশিষ্ট্য:

1. Time-based delay: প্রতিটি আইটেম নির্দিষ্ট সময়ের জন্য কিউতে থাকবে।
2. থ্রেড-সেফ: একাধিক থ্রেড একযোগে কিউ থেকে উপাদান পড়তে বা যোগ করতে পারে।
3. Delayed ইন্টারফেস: কিউর উপাদানগুলোতে getDelay() মেথড প্রয়োগ করতে হয়।

DelayQueue উদাহরণ

একটি মেসেজ ডেলিভারি সিস্টেম যেখানে মেসেজ নির্দিষ্ট সময়ে ডেলিভার হবে।

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

// Delayed ইন্টারফেস ইমপ্লিমেন্ট করা ক্লাস
class DelayedMessage implements Delayed {
    private final String message;
    private final long startTime;

    public DelayedMessage(String message, long delayInMillis) {
        this.message = message;
        this.startTime = System.currentTimeMillis() + delayInMillis;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long remainingTime = startTime - System.currentTimeMillis();
        return unit.convert(remainingTime, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if (this.startTime < ((DelayedMessage) o).startTime) {
            return -1;
        }
        if (this.startTime > ((DelayedMessage) o).startTime) {
            return 1;
        }
        return 0;
    }

    @Override
    public String toString() {
        return "Message: " + message + ", Start Time: " + startTime;
    }
}

public class DelayQueueExample {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedMessage> delayQueue = new DelayQueue<>();

        // মেসেজ যোগ করা
        delayQueue.put(new DelayedMessage("Message 1", 3000)); // ৩ সেকেন্ড বিলম্ব
        delayQueue.put(new DelayedMessage("Message 2", 5000)); // ৫ সেকেন্ড বিলম্ব

        System.out.println("Messages are being processed...");

        // মেসেজ প্রসেস করা
        while (!delayQueue.isEmpty()) {
            DelayedMessage message = delayQueue.take(); // ব্লকিং কল
            System.out.println("Processed: " + message);
        }

        System.out.println("All messages processed.");
    }
}

আউটপুট:

1. প্রথম মেসেজ ৩ সেকেন্ড পরে প্রসেস হবে।
2. দ্বিতীয় মেসেজ ৫ সেকেন্ড পরে প্রসেস হবে।

PriorityBlockingQueue

PriorityBlockingQueue হলো একটি ব্লকিং কিউ যা PriorityQueue এর থ্রেড-সেফ সংস্করণ। এটি উপাদানগুলোকে প্রাক-সংজ্ঞায়িত natural order বা কাস্টম কম্পারেটর অনুযায়ী সাজিয়ে রাখে।

বৈশিষ্ট্য:

1. Natural Ordering বা Custom Comparator: উপাদানগুলি তাদের প্রাকৃতিক অর্ডার বা নির্ধারিত কম্পারেটর অনুযায়ী সাজানো হয়।
2. থ্রেড-সেফ: একাধিক থ্রেড থেকে উপাদান যোগ বা পড়া যায়।
3. নন-ব্লকিং poll(): যদি কিউ খালি হয়, তবে poll() মেথড তাৎক্ষণিকভাবে null রিটার্ন করবে।

PriorityBlockingQueue উদাহরণ

প্রতিটি কাজকে একটি প্রায়োরিটি দিয়ে কিউতে সংরক্ষণ ও প্রসেস করা।

import java.util.concurrent.PriorityBlockingQueue;

// টাস্ক ক্লাস যার একটি প্রায়োরিটি থাকবে
class Task implements Comparable<Task> {
    private final String name;
    private final int priority;

    public Task(String name, int priority) {
        this.name = name;
        this.priority = priority;
    }

    @Override
    public int compareTo(Task other) {
        return Integer.compare(this.priority, other.priority); // Priority অনুযায়ী তুলনা
    }

    @Override
    public String toString() {
        return "Task{name='" + name + "', priority=" + priority + '}';
    }
}

public class PriorityBlockingQueueExample {
    public static void main(String[] args) throws InterruptedException {
        PriorityBlockingQueue<Task> priorityQueue = new PriorityBlockingQueue<>();

        // কাজ যোগ করা
        priorityQueue.put(new Task("Low Priority Task", 3));
        priorityQueue.put(new Task("High Priority Task", 1));
        priorityQueue.put(new Task("Medium Priority Task", 2));

        System.out.println("Tasks are being processed...");

        // কাজ প্রসেস করা
        while (!priorityQueue.isEmpty()) {
            Task task = priorityQueue.take(); // ব্লকিং কল
            System.out.println("Processed: " + task);
        }

        System.out.println("All tasks processed.");
    }
}

আউটপুট:

1. High Priority Task (priority=1) প্রথম প্রসেস হবে।
2. Medium Priority Task (priority=2) দ্বিতীয়।
3. Low Priority Task (priority=3) সর্বশেষ।

DelayQueue এবং PriorityBlockingQueue এর তুলনা

• DelayQueue: সময় ভিত্তিক কাজ বা মেসেজ ডেলিভারির জন্য ব্যবহার করুন।
• PriorityBlockingQueue: প্রায়োরিটি ভিত্তিক কাজের শিডিউলিংয়ের জন্য উপযুক্ত।

এই কিউগুলো মাল্টিথ্রেডেড পরিবেশে ডেটা ম্যানেজমেন্টকে সহজ ও কার্যকর করে।

জাভার Concurrency Utilities (যেমন ExecutorService, BlockingQueue, CountDownLatch, Semaphore, ইত্যাদি) মাল্টিথ্রেডেড প্রোগ্রামিংকে আরো সহজ, নিরাপদ এবং কার্যকর করে। তবে এগুলো সঠিকভাবে ব্যবহার না করলে ডেডলক, রেস কন্ডিশন বা পারফরম্যান্স সমস্যা দেখা দিতে পারে। নিচে Concurrency Utility ব্যবহারের জন্য কিছু Best Practices তুলে ধরা হলো।

১. ExecutorService এর ব্যবহার

ExecutorService থ্রেড ব্যবস্থাপনা সহজ করে এবং থ্রেড পুনঃব্যবহার নিশ্চিত করে।

Best Practices:

1. Fixed Thread Pool ব্যবহার করুন:

   • অপ্রয়োজনীয় থ্রেড তৈরি থেকে রক্ষা করে।

   ExecutorService executor = Executors.newFixedThreadPool(5);
   

2. সঠিকভাবে Executor বন্ধ করুন:

   • shutdown() ব্যবহার করুন কাজ শেষ হওয়ার পরে থ্রেড বন্ধ করতে।

   executor.shutdown();
   try {
       if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
           executor.shutdownNow(); // জোরপূর্বক বন্ধ
       }
   } catch (InterruptedException e) {
       executor.shutdownNow();
   }
   

3. ForkJoinPool ব্যবহার করুন বড় কাজ ভাগ করতে:

   • RecursiveTask বা RecursiveAction ব্যবহার করুন।

   ForkJoinPool pool = new ForkJoinPool();
   

২. BlockingQueue এর ব্যবহার

BlockingQueue প্রযোজক-গ্রাহক (Producer-Consumer) প্যাটার্নের জন্য অত্যন্ত কার্যকর।

Best Practices:

1. প্রযুক্তি নির্বাচন করুন:
   • কম আইটেমের জন্য ArrayBlockingQueue।
   • বড় ডেটার জন্য LinkedBlockingQueue।
   • Priority এর জন্য PriorityBlockingQueue।

2. ব্লকিং অপারেশন থেকে ডেডলক প্রতিরোধ করুন:

   • যথাসময়ে আইটেম যোগ করা বা মুছে ফেলার জন্য offer() এবং poll() ব্যবহার করুন।

   queue.offer(item, 2, TimeUnit.SECONDS);
   queue.poll(2, TimeUnit.SECONDS);
   

৩. CountDownLatch এর ব্যবহার

CountDownLatch নির্দিষ্ট সংখ্যক থ্রেড শেষ না হওয়া পর্যন্ত অপেক্ষা করতে সাহায্য করে।

Best Practices:

1. কাউন্ট সঠিকভাবে সেট করুন:

   CountDownLatch latch = new CountDownLatch(3); // তিনটি কাজের জন্য
   

2. ল্যাচ শেষ না হওয়া পর্যন্ত অপেক্ষা করুন:

   latch.await();
   

3. ল্যাচকে পুনরায় ব্যবহার না করার পরিকল্পনা করুন:
   • ল্যাচ একবার ব্যবহার করা যায়; পুনরায় ব্যবহারের জন্য CyclicBarrier ব্যবহার করুন।

৪. Semaphore এর ব্যবহার

Semaphore নির্দিষ্ট সংখ্যক থ্রেডকে রিসোর্স অ্যাক্সেসের অনুমতি দেয়।

Best Practices:

1. পারমিট সংখ্যা সঠিকভাবে নির্ধারণ করুন:

   Semaphore semaphore = new Semaphore(5); // সর্বাধিক ৫টি থ্রেড
   

2. পারমিট সঠিকভাবে মুক্ত করুন:

   • acquire() এবং release() এর জন্য চেষ্টা করুন try-finally ব্লকের মধ্যে।

   semaphore.acquire();
   try {
       // রিসোর্স অ্যাক্সেস
   } finally {
       semaphore.release();
   }
   

৫. ReadWriteLock এর ব্যবহার

ReadWriteLock একাধিক রিড থ্রেড এবং একক রাইট থ্রেডকে রিসোর্স অ্যাক্সেসের অনুমতি দেয়।

Best Practices:

1. রিড/রাইট লক সঠিকভাবে আলাদা করুন:

   ReadWriteLock lock = new ReentrantReadWriteLock();
   lock.readLock().lock();
   try {
       // রিড অপারেশন
   } finally {
       lock.readLock().unlock();
   }
   

2. ডেডলক প্রতিরোধে লক রিলিজ নিশ্চিত করুন:

   lock.writeLock().lock();
   try {
       // রাইট অপারেশন
   } finally {
       lock.writeLock().unlock();
   }
   

৬. Atomic Variables এর ব্যবহার

AtomicInteger, AtomicLong ইত্যাদি একক ভেরিয়েবলের জন্য থ্রেড-সেফ অপারেশন নিশ্চিত করে।

Best Practices:

1. কাউন্টার বা ফ্ল্যাগ ব্যবস্থাপনার জন্য ব্যবহার করুন:

   AtomicInteger counter = new AtomicInteger(0);
   counter.incrementAndGet();
   

2. CAS (Compare-And-Swap) মেকানিজমের সুবিধা নিন।

   boolean updated = atomicVar.compareAndSet(expectedValue, newValue);
   

৭. CompletableFuture এর ব্যবহার

CompletableFuture অ্যাসিনক্রোনাস টাস্ক পরিচালনার জন্য ব্যবহার করুন।

Best Practices:

1. অ্যাসিনক্রোনাস কাজ শুরুর জন্য supplyAsync() ব্যবহার করুন:

   CompletableFuture.supplyAsync(() -> "Result");
   

2. টাস্ক কম্বাইন করুন:

   CompletableFuture<String> combined = future1.thenCombine(future2, (result1, result2) -> result1 + result2);
   

3. একাধিক টাস্ক একত্রে চালান:

   CompletableFuture.allOf(future1, future2).join();
   

৮. Deadlock প্রতিরোধের কৌশল

1. কনস্ট্যান্ট লক অর্ডার ব্যবহার করুন:
   • সব লক একই সিকোয়েন্সে গ্রহণ করুন।

2. লক টাইমআউট ব্যবহার করুন:

   if (lock.tryLock(5, TimeUnit.SECONDS)) {
       try {
           // কাজ
       } finally {
           lock.unlock();
       }
   }
   

৯. Logging এবং Monitoring

• লগিং: থ্রেড ডাম্প এবং কনকারেন্সি সম্পর্কিত ত্রুটি লগ করুন।
• মন্টরিং টুল: JVisualVM বা JConsole ব্যবহার করে থ্রেডের কার্যকলাপ পর্যবেক্ষণ করুন।

Concurrency Utility গুলোর জন্য Best Practices অনুসরণ করলে:

1. সঠিক থ্রেড ব্যবস্থাপনা: ডেডলক, রেস কন্ডিশন এড়ানো যাবে।
2. উন্নত পারফরম্যান্স: রিসোর্সের কার্যকর ব্যবহার নিশ্চিত হবে।
3. সহজ রক্ষণাবেক্ষণ: কোড আরো পড়তে এবং রক্ষণাবেক্ষণে সহজ হবে।

এই প্র্যাকটিসগুলো জাভার মাল্টিথ্রেডেড প্রোগ্রামিংয়ে কনকারেন্সি সমস্যাগুলোর সমাধান নিশ্চিত করবে।