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}} 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

Processor Run Method

Now, we’ll implement the run method for the Processor to handle taking, processing, and passing data through the queue.

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
    }
}

The Processor continuously takes items from the input queue and processes them, managing concurrency with a semaphore. Upon receiving the poison pill, it forwards it to the next stage and completes its work.

Putting It All Together - Main Class

Now that we've implemented the Producer, Processor, and Consumer, we can set up the Main class to run the system and tie everything together.

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.");
    }
}

The Main class brings all the components together, setting up the pipeline of producer, processor, and consumer with the necessary synchronization mechanisms. The Phaser ensures that all threads complete before the program exits, and the Semaphore limits the number of concurrent processors.

With the main class implemented, the entire system is now functional.

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