cyclic barrier , synchronized queues

Author: Ashfaqbs

.idea/.gitignore

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>Concurrency-Multithreading-and-Parallel-Computing-in-Java</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+	</buildSpec>
+	<natures>
+	</natures>
+</projectDescription>

.idea/Concurrency-Multithreading-and-Parallel-Computing-in-Java.iml

@@ -0,0 +1,74 @@
+package org.ashfaq.dev.concepts.Concurrent_Collections;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+
+//A BlockingQueue is a type of queue that supports operations that wait for the queue to become non-empty when retrieving
+//an element and wait for space
+//to become available in the queue when storing an element. This is particularly useful in a producer-consumer scenario.
+class Producer implements Runnable {
+    private final BlockingQueue<Integer> queue;
+
+    public Producer(BlockingQueue<Integer> queue) {
+        this.queue = queue;
+    }
+
+    @Override
+    public void run() {
+        try {
+            int value = 0;
+            while (true) {
+                System.out.println("Producing " + value);
+                queue.put(value);
+                value++;
+                Thread.sleep(1000); // Simulate time taken to produce
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            System.out.println("Producer interrupted.");
+        }
+    }
+}
+
+class Consumer implements Runnable {
+    private final BlockingQueue<Integer> queue;
+
+    public Consumer(BlockingQueue<Integer> queue) {
+        this.queue = queue;
+    }
+
+    @Override
+    public void run() {
+        try {
+            while (true) {
+                Integer value = queue.take();
+                System.out.println("Consuming " + value);
+                Thread.sleep(1500); // Simulate time taken to consume
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            System.out.println("Consumer interrupted.");
+        }
+    }
+}
+
+public class BlockingQueueExample {
+    public static void main(String[] args) {
+        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+
+        executor.submit(new Producer(queue));
+        executor.submit(new Consumer(queue));
+
+        // Shutdown the executor after some time for demonstration
+        try {
+            Thread.sleep(10000);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+
+        executor.shutdownNow();
+    }
+}

.idea/misc.xml

@@ -0,0 +1,85 @@
+package org.ashfaq.dev.concepts.Concurrent_Collections;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+class Task implements Runnable {
+    private final CyclicBarrier barrier;
+    private final int id;
+
+    public Task(CyclicBarrier barrier, int id) {
+        this.barrier = barrier;
+        this.id = id;
+    }
+
+    @Override
+    public void run() {
+        try {
+            System.out.println("Task " + id + " is performing part 1.");
+            Thread.sleep((int) (Math.random() * 1000));
+            System.out.println("Task " + id + " reached the barrier.");
+            barrier.await(); // Wait for other threads
+            System.out.println("Task " + id + " is performing part 2.");
+        } catch (InterruptedException | BrokenBarrierException e) {
+            e.printStackTrace();
+        }
+    }
+}
+
+public class CyclicBarrierExample {
+    public static void main(String[] args) {
+        // Create a CyclicBarrier for 3 threads with a barrier action
+        CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("All tasks have reached the barrier. Proceeding..."));
+        
+
+        // Create and start three tasks
+//        Thread t1 = new Thread(new Task(barrier, 1));
+//        Thread t2 = new Thread(new Task(barrier, 2));
+//        Thread t3 = new Thread(new Task(barrier, 3));
+//
+//        t1.start();
+//        t2.start();
+//        t3.start();
+//        OP
+        //Task 2 is performing part 1.
+//        Task 1 is performing part 1.
+//        Task 3 is performing part 1.
+//        Task 2 reached the barrier.
+//        Task 1 reached the barrier.
+//        Task 3 reached the barrier.
+//        All tasks have reached the barrier. Proceeding...
+//        Task 2 is performing part 2.
+//        Task 1 is performing part 2.
+//        Task 3 is performing part 2.
+        
+        
+//        Example Using Executors and CyclicBarrier
+
+        
+        int numberOfThreads = 3;
+        // Create an ExecutorService with a fixed thread pool
+        ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
+
+        // Submit tasks to the executor service
+        for (int i = 1; i <= numberOfThreads; i++) {
+            executorService.submit(new Task(barrier, i));
+        }
+
+        // Shutdown the executor service
+        executorService.shutdown();
+        
+//        OP
+//        Task 2 is performing part 1.
+//        Task 3 is performing part 1.
+//        Task 1 is performing part 1.
+//        Task 1 reached the barrier.
+//        Task 3 reached the barrier.
+//        Task 2 reached the barrier.
+//        All tasks have reached the barrier. Proceeding...
+//        Task 3 is performing part 2.
+//        Task 2 is performing part 2.
+//        Task 1 is performing part 2.
+    }
+}

.idea/modules.xml

@@ -0,0 +1,73 @@
+package org.ashfaq.dev.concepts.Concurrent_Collections;
+
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+class DelayedTask implements Delayed {
+    private final String name;
+    private final long delayTime;
+    private final long expireTime;
+
+    public DelayedTask(String name, long delayTime) {
+        this.name = name;
+        this.delayTime = delayTime;
+        this.expireTime = System.currentTimeMillis() + delayTime;
+    }
+
+    @Override
+    public long getDelay(TimeUnit unit) {
+        long diff = expireTime - System.currentTimeMillis();
+        return unit.convert(diff, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(Delayed o) {
+        if (this.expireTime < ((DelayedTask) o).expireTime) {
+            return -1;
+        }
+        if (this.expireTime > ((DelayedTask) o).expireTime) {
+            return 1;
+        }
+        return 0;
+    }
+
+    public void execute() {
+        System.out.println("Executing task: " + name + " at " + System.currentTimeMillis());
+    }
+}
+
+public class DelayQueueExample {
+    public static void main(String[] args) {
+        DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
+        ExecutorService executor = Executors.newFixedThreadPool(1);
+
+        // Add tasks to the delay queue
+        delayQueue.put(new DelayedTask("Task1", 5000)); // 5 seconds delay
+        delayQueue.put(new DelayedTask("Task2", 3000)); // 3 seconds delay
+        delayQueue.put(new DelayedTask("Task3", 10000)); // 10 seconds delay
+
+        // Consumer thread to execute tasks after their delay
+        executor.submit(() -> {
+            while (!Thread.currentThread().isInterrupted()) {
+                try {
+                    DelayedTask task = delayQueue.take(); // Blocks until a task's delay has expired
+                    task.execute();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+        });
+
+        // For demonstration, we'll shut down the executor after some time
+        try {
+            Thread.sleep(15000); // Let it run for 15 seconds
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+
+        executor.shutdownNow();
+    }
+}

.idea/vcs.xml

@@ -0,0 +1,82 @@
+package org.ashfaq.dev.concepts.Concurrent_Collections;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
+
+class FirstWorker implements Runnable {
+
+	private BlockingQueue<String> queue;
+
+	public FirstWorker(BlockingQueue<String> queue) {
+		this.queue = queue;
+	}
+
+	@Override
+	public void run() {
+		try {
+
+			queue.put("A");
+			queue.put("B");
+			queue.put("C");
+			Thread.sleep(2000);
+			queue.put("D");
+			Thread.sleep(2000);
+			queue.put("E");
+			queue.put("F");
+
+		} catch (InterruptedException e) {
+		}
+	}
+}
+
+class SecondWorker implements Runnable {
+
+	private BlockingQueue<String> queue;
+
+	public SecondWorker(BlockingQueue<String> queue) {
+		this.queue = queue;
+	}
+
+	@Override
+	public void run() {
+		try {
+
+			Thread.sleep(5000);
+			System.out.println(queue.take());
+			Thread.sleep(2000);
+			System.out.println(queue.take());
+			Thread.sleep(2000);
+			System.out.println(queue.take());
+			System.out.println(queue.take());
+			System.out.println(queue.take());
+			System.out.println(queue.take());
+
+		} catch (InterruptedException e) {
+		}
+	}
+}
+
+
+public class PriorityQueue {
+
+	public static void main(String[] args) {
+
+		BlockingQueue<String> queue = new PriorityBlockingQueue<>();
+		FirstWorker firstWorker = new FirstWorker(queue);
+		SecondWorker secondWorker = new SecondWorker(queue);
+
+		Thread t1 = new Thread(firstWorker);
+		Thread t2 = new Thread(secondWorker);
+
+		t1.start();
+		t2.start();
+		
+		
+		
+		
+		
+		
+
+	}
+
+}