Building a Producer-Consumer System with BlockingQueue

Hello again! In our previous lessons, we explored using various concurrent collections like ConcurrentHashMap and CopyOnWriteArrayList to manage data across multiple threads safely. Today, we will dive into a classic concurrency pattern: the Producer-Consumer system, utilizing BlockingQueue . This will further enhance your understanding of managing concurrent tasks efficiently.

What You'll Learn

In this session, you'll acquire the skills to: Implement a Producer-Consumer pattern using BlockingQueue . Understand how BlockingQueue handles data exchange between threads. Manage thread communication effectively in concurrent applications. These are crucial skills for developing systems that process tasks asynchronously, such as job scheduling or handling requests on a server.

Recap: Producer-Consumer Pattern

Before diving into the new lesson, let’s recap the Producer-Consumer pattern—a concurrency model we have covered in previous lessons. The Producer-Consumer pattern is a classic problem where: Producers are threads that generate data (or tasks), typically at varying rates. Consumers are threads that process or consume this data. The key challenge is coordinating these two types of threads so that producers don't overwhelm consumers with too much data and consumers don't run out of data to process.

The Role of BlockingQueue in Producer-Consumer Systems

In this pattern, a BlockingQueue acts as a buffer or intermediary for data between the producers and consumers. The BlockingQueue solves several problems at once: Thread Safety : It is designed to handle multiple producers and consumers working concurrently, without requiring manual synchronization. Blocking Behavior : Producers use methods like put(), which block if the queue is full, ensuring that they don’t overload the system with too much data. Consumers use methods like take(), which block if the queue is empty, ensuring that they don’t consume non-existent data or busy-wait for the producer to create tasks. This blocking behavior helps balance the workload between producers and consumers, making it an efficient solution for concurrent task processing.

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;

public class Producer implements Runnable {
    private final BlockingQueue<String> queue;
    private final int messageCount;

    public Producer(BlockingQueue<String> queue, int messageCount) {
        this.queue = queue;
        this.messageCount = messageCount;
    }
}

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {
    private final BlockingQueue<String> queue;
    private final int messageCount;

    public Producer(BlockingQueue<String> queue, int messageCount) {
        this.queue = queue;
        this.messageCount = messageCount;
    }
}

Java

@Override
    public void run() {
        try {
            for (int i = 1; i <= messageCount; i++) {
                String message = "Message " + i;
                queue.put(message); // Blocks if the queue is full
                System.out.println(Thread.currentThread().getName() + " produced: " + message);
                Thread.sleep(100); // Simulate production time
            }
            // Indicate end of production
            queue.put("DONE");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Producer interrupted.");
        }
    }

@Override
    public void run() {
        try {
            for (int i = 1; i <= messageCount; i++) {
                String message = "Message " + i;
                queue.put(message); // Blocks if the queue is full
                System.out.println(Thread.currentThread().getName() + " produced: " + message);
                Thread.sleep(100); // Simulate production time
            }
            // Indicate end of production
            queue.put("DONE");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Producer interrupted.");
        }
    }

Java

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {
    private final BlockingQueue<String> queue;

    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }
}

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {
    private final BlockingQueue<String> queue;

    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }
}

Java

@Override
    public void run() {
        try {
            while (true) {
                String message = queue.take(); // Blocks if the queue is empty
                if ("DONE".equals(message)) {
                    System.out.println(Thread.currentThread().getName() + " received DONE signal.");
                    break;
                }
                System.out.println(Thread.currentThread().getName() + " consumed: " + message);
                Thread.sleep(150); // Simulate processing time
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Consumer interrupted.");
        }
    }

@Override
    public void run() {
        try {
            while (true) {
                String message = queue.take(); // Blocks if the queue is empty
                if ("DONE".equals(message)) {
                    System.out.println(Thread.currentThread().getName() + " received DONE signal.");
                    break;
                }
                System.out.println(Thread.currentThread().getName() + " consumed: " + message);
                Thread.sleep(150); // Simulate processing time
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Consumer interrupted.");
        }
    }

Java

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

public class Main {
    public static void main(String[] args) {
        // Initialize a LinkedBlockingQueue with a capacity of 5
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(5);

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

public class Main {
    public static void main(String[] args) {
        // Initialize a LinkedBlockingQueue with a capacity of 5
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(5);

Java

        // Create Producer and Consumer instances
        Producer producer = new Producer(queue, 10);
        Consumer consumer = new Consumer(queue);

        // Initialize Threads
        Thread producerThread = new Thread(producer, "Producer-Thread");
        Thread consumerThread = new Thread(consumer, "Consumer-Thread");

        // Start Threads
        producerThread.start();
        consumerThread.start();

        // Create Producer and Consumer instances
        Producer producer = new Producer(queue, 10);
        Consumer consumer = new Consumer(queue);

        // Initialize Threads
        Thread producerThread = new Thread(producer, "Producer-Thread");
        Thread consumerThread = new Thread(consumer, "Consumer-Thread");

        // Start Threads
        producerThread.start();
        consumerThread.start();

Java

        // Wait for Threads to Finish
        try {
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Main thread interrupted.");
        }

        System.out.println("Producer-Consumer execution completed.");
    }
}

        // Wait for Threads to Finish
        try {
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Main thread interrupted.");
        }

        System.out.println("Producer-Consumer execution completed.");
    }
}

Java

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {
    private final BlockingQueue<String> queue;
    private final int messageCount;

    public Producer(BlockingQueue<String> queue, int messageCount) {
        this.queue = queue;
        this.messageCount = messageCount;
    }
}

Java

@Override
    public void run() {
        try {
            for (int i = 1; i <= messageCount; i++) {
                String message = "Message " + i;
                queue.put(message); // Blocks if the queue is full
                System.out.println(Thread.currentThread().getName() + " produced: " + message);
                Thread.sleep(100); // Simulate production time
            }
            // Indicate end of production
            queue.put("DONE");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Producer interrupted.");
        }
    }

Java

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {
    private final BlockingQueue<String> queue;

    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }
}

Java

@Override
    public void run() {
        try {
            while (true) {
                String message = queue.take(); // Blocks if the queue is empty
                if ("DONE".equals(message)) {
                    System.out.println(Thread.currentThread().getName() + " received DONE signal.");
                    break;
                }
                System.out.println(Thread.currentThread().getName() + " consumed: " + message);
                Thread.sleep(150); // Simulate processing time
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Consumer interrupted.");
        }
    }

Java

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

public class Main {
    public static void main(String[] args) {
        // Initialize a LinkedBlockingQueue with a capacity of 5
        BlockingQueue<String> queue = new LinkedBlockingQueue<>(5);

Java

        // Create Producer and Consumer instances
        Producer producer = new Producer(queue, 10);
        Consumer consumer = new Consumer(queue);

        // Initialize Threads
        Thread producerThread = new Thread(producer, "Producer-Thread");
        Thread consumerThread = new Thread(consumer, "Consumer-Thread");

        // Start Threads
        producerThread.start();
        consumerThread.start();

Java

        // Wait for Threads to Finish
        try {
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Main thread interrupted.");
        }

        System.out.println("Producer-Consumer execution completed.");
    }
}

Building a Producer-Consumer System with BlockingQueue

What You'll Learn

Recap: Producer-Consumer Pattern

The Role of BlockingQueue in Producer-Consumer Systems

Join the 1M+ learners on CodeSignal

Recap: BlockingQueue and Its Functionality

Implementing the Producer Class

Producer's run() Method

Implementing the Consumer Class

Consumer's run() Method

Running the Producer-Consumer System

Creating and Starting Threads

Waiting for Threads to Finish

Why It Matters