Blocking Queues

//LinkedBlockingQueueDemo.java
import java.util.concurrent.LinkedBlockingQueue;
public class LinkedBlockingQueueDemo {

    // Define the size of the blocking queue
    private static final int QUEUE_CAPACITY = 10;

    // Create the shared blocking queue
    private static final LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);

    public static void main(String[] args) {
        // Create and start producer and consumer threads
        Thread producerThread = new Thread(new Producer(queue));
        Thread consumerThread = new Thread(new Consumer(queue));

        producerThread.start();
        consumerThread.start();
    }

    // Producer class that puts items into the queue
    static class Producer implements Runnable {
        private final LinkedBlockingQueue<Integer> queue;
        private int item = 0;

        public Producer(LinkedBlockingQueue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    int producedItem = produce();
                    queue.put(producedItem);  // Add item to queue, blocks if queue is full
                    System.out.println("Produced: " + producedItem);

                    // Notify consumers if needed - not strictly necessary with LinkedBlockingQueue
                    synchronized (queue) {
                        queue.notifyAll();
                    }

                    Thread.sleep(500);  // Simulate time taken to produce an item
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        private int produce() {
            return item++;
        }
    }

    // Consumer class that takes items from the queue
    static class Consumer implements Runnable {
        private final LinkedBlockingQueue<Integer> queue;

        public Consumer(LinkedBlockingQueue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    Integer item = queue.take();  // Remove item from queue, blocks if queue is empty
                    System.out.println("Consumed: " + item);
                    process(item);

                    // Notify producer if needed - not strictly necessary with LinkedBlockingQueue
                    synchronized (queue) {
                        queue.notifyAll();
                    }

                    Thread.sleep(1000);  // Simulate time taken to consume an item
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        private void process(Integer item) {
            // Simulate processing the item
            System.out.println("Processing item: " + item);
        }
    }
}
/*
:\Users\AITS_CCF\Desktop\Bhava Advanced Concurrency\Bhava Edited\Thread Communication\04>javac LinkedBlockingQueueDemo.java

C:\Users\AITS_CCF\Desktop\Bhava Advanced Concurrency\Bhava Edited\Thread Communication\04>java LinkedBlockingQueueDemo
Consumed: 0
Processing item: 0
Produced: 0
Produced: 1
Consumed: 1
Processing item: 1
Produced: 2
Produced: 3
Consumed: 2
Processing item: 2
Produced: 4
Produced: 5
Consumed: 3
Processing item: 3
Produced: 6
Produced: 7
Consumed: 4
Processing item: 4
Produced: 8
Produced: 9
Consumed: 5
Processing item: 5
Produced: 10
Produced: 11
Consumed: 6
Processing item: 6
Produced: 12
Produced: 13
Consumed: 7
Processing item: 7
Produced: 14
Produced: 15
Consumed: 8
Processing item: 8
Produced: 16
Produced: 17
Consumed: 9
Processing item: 9
Produced: 18
Produced: 19
Consumed: 10
Processing item: 10
Produced: 20
Consumed: 11
Produced: 21
Processing item: 11
Consumed: 12
Produced: 22
Processing item: 12
Produced: 23
Consumed: 13
Processing item: 13
Consumed: 14
Produced: 24
Processing item: 14
Produced: 25
Consumed: 15
Processing item: 15
Consumed: 16
Produced: 26
Processing item: 16
Consumed: 17
Produced: 27
Processing item: 17
Consumed: 18
Produced: 28
*/

//PriorityBlockingQueueDemo.java
import java.util.concurrent.PriorityBlockingQueue;

class Task implements Comparable<Task> {
    private String name;
    private int priority;
    
    // Special poison pill task to signal consumer to stop
    public static final Task POISON_PILL = new Task("POISON_PILL", Integer.MAX_VALUE);

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

    public String getName() {
        return name;
    }

    public int getPriority() {
        return priority;
    }

    @Override
    public int compareTo(Task other) {
        return Integer.compare(this.priority, other.priority);
    }
}

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

        // Producer thread adding tasks with different priorities
        Thread producer = new Thread(() -> {
            try {
                queue.put(new Task("Task 1", 3)); // Lowest priority
                queue.put(new Task("Task 2", 1)); // Highest priority
                queue.put(new Task("Task 3", 2)); // Medium priority
                queue.put(Task.POISON_PILL); // Poison pill to signal consumer to stop
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        // Consumer thread processing tasks in priority order
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    Task task = queue.take(); // Block until a task is available
                    if (task == Task.POISON_PILL) {
                        System.out.println("Received poison pill. Consumer is shutting down.");
                        break; // Exit loop once poison pill is received
                    }
                    System.out.println("Processing task: " + task.getName() + " with priority " + task.getPriority());
                    Thread.sleep(1000); // Simulate processing time
                }
            } catch (InterruptedException e) {
                System.out.println("Consumer thread interrupted.");
            }
        });

        producer.start();
        consumer.start();

        // Let the producer and consumer threads finish
        producer.join();
        consumer.join();
    }
}

/*

C:\>javac PriorityBlockingQueueDemo.java

C:\>java PriorityBlockingQueueDemo
Processing task: Task 2 with priority 1
Processing task: Task 3 with priority 2
Processing task: Task 1 with priority 3
Received poison pill. Consumer is shutting down.
*/

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

// Example class representing a delayed element
class DelayedElement implements Delayed {
    private String name;
    private long delayTime; // Delay period in milliseconds
    private long expireTime; // Expiration time in milliseconds

    public DelayedElement(String name, long delayTime) {
        this.name = name;
        this.delayTime = delayTime;
        this.expireTime = System.currentTimeMillis() + delayTime;
    }

    public String getName() {
        return name;
    }

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

    @Override
    public int compareTo(Delayed other) {
        long diff = this.getDelay(TimeUnit.MILLISECONDS) - other.getDelay(TimeUnit.MILLISECONDS);
        return Long.compare(diff, 0);
    }
}

public class DelayQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedElement> queue = new DelayQueue<>();

        // Producer thread adding delayed elements
        Thread producer = new Thread(() -> {
            try {
                queue.put(new DelayedElement("Delayed Element 1", 2000)); // Delay of 2 seconds
                queue.put(new DelayedElement("Delayed Element 2", 4000)); // Delay of 4 seconds
                queue.put(new DelayedElement("Delayed Element 3", 6000)); // Delay of 6 seconds
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        // Consumer thread processing delayed elements
        Thread consumer = new Thread(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    DelayedElement element = queue.take(); // This will block until a delayed element's delay has expired
                    System.out.println("Processing delayed element: " + element.getName());
                }
            } catch (InterruptedException e) {
                System.out.println("Consumer thread interrupted, exiting...");
            }
        });

        producer.start();
        consumer.start();

        // Let the consumer run for some time
        Thread.sleep(8000);
        consumer.interrupt(); // Interrupt the consumer to stop the example gracefully

        producer.join();
        consumer.join();
    }
}


/*
C:\>javac DelayQueueDemo.java

C:\>java DelayQueueDemo
Processing delayed element: Delayed Element 1
Processing delayed element: Delayed Element 2
Processing delayed element: Delayed Element 3
Consumer thread interrupted, exiting...

*/

import java.util.concurrent.SynchronousQueue;
//SynchronousQueueDemo.java
public class SynchronousQueueDemo{
    public static void main(String[] args) throws InterruptedException {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();

        // Producer thread putting data into the queue
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    System.out.println("Producing: " + i);
                    queue.put(i); // This will block until a consumer takes the data
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        // Consumer thread taking data from the queue
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    int value = queue.take(); // This will block until a producer puts data
                    System.out.println("Consuming: " + value);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();
    }
}

/*
C:\>javac SynchronousQueueDemo.java

C:\>java SynchronousQueueDemo
Producing: 1
Producing: 2
Consuming: 1
Producing: 3
Consuming: 2
Producing: 4
Consuming: 3
Producing: 5
Consuming: 4
Consuming: 5
*/