Applying Advanced Concurrency Utilities to Practical Scenarios

Applying Advanced Utilities to Real-Life Problems

Welcome back to our deep dive into advanced concurrency utilities! Previously, we explored various Java synchronization tools like Semaphores, CyclicBarriers, Phasers, and the Exchanger . These tools help manage complex interactions between threads, enhancing your ability to tackle concurrency challenges effectively. Today, we will demonstrate how these elements come together to solve real-life concurrency problems. This lesson builds on what you've learned so far, emphasizing practical applications of these utilities. Let’s dive into a scenario that integrates these advanced tools.

What You'll Learn

By the end of this lesson, you will: Understand how to use multiple concurrency utilities together to solve a complex problem. Gain insights into designing a comprehensive solution using Semaphore, Phaser, and BlockingQueue . Explore the application of these tools in a real-life data processing workflow, learning how concurrency tools can optimize resource management and synchronization. Through these objectives, you will gain a deeper understanding of how to solve concurrency-related challenges using these utilities.

Understanding the Application of Advanced Utilities

In this lesson, we’ll integrate several concurrency utilities into a cohesive solution, simulating a data processing pipeline. We use a BlockingQueue to manage data flow between stages, a Semaphore to control the number of concurrent threads during processing, and a Phaser to synchronize the threads at the end of the pipeline's operation. The scenario involves a producer generating items, a processor modifying these items, and a consumer that consumes them. A poison pill is utilized to signal the end of data flow through the system. This setup mirrors real-world scenarios like algorithms, data pipelines, or multithreaded server processing, where coordinated, efficient resource management is crucial for throughput and stability. Now, let’s break down the implementation of each part of this pipeline.

Producer Implementation

Let’s start by implementing the Producer, which will generate items and place them in the first BlockingQueue . The producer will use a Phaser to notify when it's done, and it will send a poison pill at the end to signal the termination of the pipeline. Java import java.util.concurrent.BlockingQueue; import java.util.concurrent.Phaser; public class Producer implements Runnable {private BlockingQueue<Integer> outputQueue; private int itemCount; private int poisonPill; private Phaser phaser; public Producer(BlockingQueue<Integer> outputQueue, int itemCount, int poisonPill, Phaser phaser) {this.outputQueue = outputQueue; this.itemCount = itemCount; this.poisonPill = poisonPill; this.phaser = phaser; this.phaser.register(); // Register this thread with the phaser}} import java.util.concurrent.BlockingQueue; import java.util.concurrent.Phaser; public class Producer implements Runnable {private BlockingQueue<Integer> outputQueue; private int itemCount; private int poisonPill; private Phaser phaser; public Producer(BlockingQueue<Integer> outputQueue, int itemCount, int poisonPill, Phaser phaser) {this.outputQueue = outputQueue; this.itemCount = itemCount; this.poisonPill = poisonPill; this.phaser = phaser; this.phaser.register(); // Register this thread with the phaser}} This sets up the producer's main function, which will be explained further in the following section.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Phaser;

public class Producer implements Runnable {
    private BlockingQueue<Integer> outputQueue;
    private int itemCount;
    private int poisonPill;
    private Phaser phaser;

    public Producer(BlockingQueue<Integer> outputQueue, int itemCount, int poisonPill, Phaser phaser) {
        this.outputQueue = outputQueue;
        this.itemCount = itemCount;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

Java

@Override
public void run() {
    try {
        for (int i = 1; i <= itemCount; i++) {
            outputQueue.put(i); // Produce data
            System.out.println("Produced: " + i);
        }
        outputQueue.put(poisonPill); // Signal completion to the next stage
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

@Override
public void run() {
    try {
        for (int i = 1; i <= itemCount; i++) {
            outputQueue.put(i); // Produce data
            System.out.println("Produced: " + i);
        }
        outputQueue.put(poisonPill); // Signal completion to the next stage
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

Java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.Phaser;

public class Processor implements Runnable {
    private BlockingQueue<Integer> inputQueue;
    private BlockingQueue<Integer> outputQueue;
    private Semaphore semaphore;
    private int poisonPill;
    private Phaser phaser;

    public Processor(BlockingQueue<Integer> inputQueue, BlockingQueue<Integer> outputQueue, Semaphore semaphore, int poisonPill, Phaser phaser) {
        this.inputQueue = inputQueue;
        this.outputQueue = outputQueue;
        this.semaphore = semaphore;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.Phaser;

public class Processor implements Runnable {
    private BlockingQueue<Integer> inputQueue;
    private BlockingQueue<Integer> outputQueue;
    private Semaphore semaphore;
    private int poisonPill;
    private Phaser phaser;

    public Processor(BlockingQueue<Integer> inputQueue, BlockingQueue<Integer> outputQueue, Semaphore semaphore, int poisonPill, Phaser phaser) {
        this.inputQueue = inputQueue;
        this.outputQueue = outputQueue;
        this.semaphore = semaphore;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

Java

@Override
public void run() {
    try {
        while (true) {
            Integer item = inputQueue.take(); // Take item from input queue
            if (item.equals(poisonPill)) {
                outputQueue.put(poisonPill); // Pass poison pill downstream
                break; // Exit loop
            }
            semaphore.acquire(); // Acquire permit to process
            System.out.println("Processing: " + item);

            // Simulate processing (e.g., doubling the number)
            int processedItem = item * 2;

            outputQueue.put(processedItem); // Put processed item into output queue
            System.out.println("Processed: " + processedItem);
            semaphore.release(); // Release permit
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

@Override
public void run() {
    try {
        while (true) {
            Integer item = inputQueue.take(); // Take item from input queue
            if (item.equals(poisonPill)) {
                outputQueue.put(poisonPill); // Pass poison pill downstream
                break; // Exit loop
            }
            semaphore.acquire(); // Acquire permit to process
            System.out.println("Processing: " + item);

            // Simulate processing (e.g., doubling the number)
            int processedItem = item * 2;

            outputQueue.put(processedItem); // Put processed item into output queue
            System.out.println("Processed: " + processedItem);
            semaphore.release(); // Release permit
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

Java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Phaser;

public class Consumer implements Runnable {
    private BlockingQueue<Integer> inputQueue;
    private int poisonPill;
    private Phaser phaser;

    public Consumer(BlockingQueue<Integer> inputQueue, int poisonPill, Phaser phaser) {
        this.inputQueue = inputQueue;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Phaser;

public class Consumer implements Runnable {
    private BlockingQueue<Integer> inputQueue;
    private int poisonPill;
    private Phaser phaser;

    public Consumer(BlockingQueue<Integer> inputQueue, int poisonPill, Phaser phaser) {
        this.inputQueue = inputQueue;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

Java

@Override
public void run() {
    try {
        while (true) {
            Integer item = inputQueue.take(); // Take item from input queue
            if (item.equals(poisonPill)) {
                break; // Exit loop
            }
            System.out.println("Consumed: " + item);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

@Override
public void run() {
    try {
        while (true) {
            Integer item = inputQueue.take(); // Take item from input queue
            if (item.equals(poisonPill)) {
                break; // Exit loop
            }
            System.out.println("Consumed: " + item);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

Java

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

public class Main {
    public static void main(String[] args) {
        final int ITEM_COUNT = 10; // Number of items to process
        final int POISON_PILL = -1; // Special value to signal completion

        // BlockingQueues between stages
        BlockingQueue<Integer> queue1 = new LinkedBlockingQueue<>();
        BlockingQueue<Integer> queue2 = new LinkedBlockingQueue<>();

        // Semaphore to limit processing concurrency
        Semaphore semaphore = new Semaphore(3); // Limit to 3 concurrent processing threads

        // Phaser for synchronizing completion of threads
        Phaser phaser = new Phaser(1); // Main thread registered

        // Create and start threads for each stage
        Thread producerThread = new Thread(new Producer(queue1, ITEM_COUNT, POISON_PILL, phaser));
        Thread processorThread = new Thread(new Processor(queue1, queue2, semaphore, POISON_PILL, phaser));
        Thread consumerThread = new Thread(new Consumer(queue2, POISON_PILL, phaser));

        producerThread.start();
        processorThread.start();
        consumerThread.start();

        // Wait for all threads to complete
        phaser.arriveAndAwaitAdvance();

        System.out.println("Processing complete.");
    }
}

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

public class Main {
    public static void main(String[] args) {
        final int ITEM_COUNT = 10; // Number of items to process
        final int POISON_PILL = -1; // Special value to signal completion

        // BlockingQueues between stages
        BlockingQueue<Integer> queue1 = new LinkedBlockingQueue<>();
        BlockingQueue<Integer> queue2 = new LinkedBlockingQueue<>();

        // Semaphore to limit processing concurrency
        Semaphore semaphore = new Semaphore(3); // Limit to 3 concurrent processing threads

        // Phaser for synchronizing completion of threads
        Phaser phaser = new Phaser(1); // Main thread registered

        // Create and start threads for each stage
        Thread producerThread = new Thread(new Producer(queue1, ITEM_COUNT, POISON_PILL, phaser));
        Thread processorThread = new Thread(new Processor(queue1, queue2, semaphore, POISON_PILL, phaser));
        Thread consumerThread = new Thread(new Consumer(queue2, POISON_PILL, phaser));

        producerThread.start();
        processorThread.start();
        consumerThread.start();

        // Wait for all threads to complete
        phaser.arriveAndAwaitAdvance();

        System.out.println("Processing complete.");
    }
}

Java

@Override
public void run() {
    try {
        for (int i = 1; i <= itemCount; i++) {
            outputQueue.put(i); // Produce data
            System.out.println("Produced: " + i);
        }
        outputQueue.put(poisonPill); // Signal completion to the next stage
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

Java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.Phaser;

public class Processor implements Runnable {
    private BlockingQueue<Integer> inputQueue;
    private BlockingQueue<Integer> outputQueue;
    private Semaphore semaphore;
    private int poisonPill;
    private Phaser phaser;

    public Processor(BlockingQueue<Integer> inputQueue, BlockingQueue<Integer> outputQueue, Semaphore semaphore, int poisonPill, Phaser phaser) {
        this.inputQueue = inputQueue;
        this.outputQueue = outputQueue;
        this.semaphore = semaphore;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

Java

@Override
public void run() {
    try {
        while (true) {
            Integer item = inputQueue.take(); // Take item from input queue
            if (item.equals(poisonPill)) {
                outputQueue.put(poisonPill); // Pass poison pill downstream
                break; // Exit loop
            }
            semaphore.acquire(); // Acquire permit to process
            System.out.println("Processing: " + item);

            // Simulate processing (e.g., doubling the number)
            int processedItem = item * 2;

            outputQueue.put(processedItem); // Put processed item into output queue
            System.out.println("Processed: " + processedItem);
            semaphore.release(); // Release permit
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

Java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Phaser;

public class Consumer implements Runnable {
    private BlockingQueue<Integer> inputQueue;
    private int poisonPill;
    private Phaser phaser;

    public Consumer(BlockingQueue<Integer> inputQueue, int poisonPill, Phaser phaser) {
        this.inputQueue = inputQueue;
        this.poisonPill = poisonPill;
        this.phaser = phaser;
        this.phaser.register(); // Register this thread with the phaser
    }
}

Java

@Override
public void run() {
    try {
        while (true) {
            Integer item = inputQueue.take(); // Take item from input queue
            if (item.equals(poisonPill)) {
                break; // Exit loop
            }
            System.out.println("Consumed: " + item);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        phaser.arriveAndDeregister(); // Signal completion to the phaser
    }
}

Java

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

public class Main {
    public static void main(String[] args) {
        final int ITEM_COUNT = 10; // Number of items to process
        final int POISON_PILL = -1; // Special value to signal completion

        // BlockingQueues between stages
        BlockingQueue<Integer> queue1 = new LinkedBlockingQueue<>();
        BlockingQueue<Integer> queue2 = new LinkedBlockingQueue<>();

        // Semaphore to limit processing concurrency
        Semaphore semaphore = new Semaphore(3); // Limit to 3 concurrent processing threads

        // Phaser for synchronizing completion of threads
        Phaser phaser = new Phaser(1); // Main thread registered

        // Create and start threads for each stage
        Thread producerThread = new Thread(new Producer(queue1, ITEM_COUNT, POISON_PILL, phaser));
        Thread processorThread = new Thread(new Processor(queue1, queue2, semaphore, POISON_PILL, phaser));
        Thread consumerThread = new Thread(new Consumer(queue2, POISON_PILL, phaser));

        producerThread.start();
        processorThread.start();
        consumerThread.start();

        // Wait for all threads to complete
        phaser.arriveAndAwaitAdvance();

        System.out.println("Processing complete.");
    }
}

Applying Advanced Utilities to Real-Life Problems

What You'll Learn

Understanding the Application of Advanced Utilities

Producer Implementation

Join the 1M+ learners on CodeSignal

Producer Run Method

Processor Implementation

Processor Run Method

Consumer Implementation

Consumer Run Method

Putting It All Together - Main Class

Why It Matters