Java Producer - Consumer Examples Using BlockingQueue
- Details
- Written by Nam Ha Minh
- Last Updated on 16 June 2019 | Print Email
- Chat applications: Messages are put into a queue. When you are sending a message, you are the producer; and your friend who reads the message, is the consumer. Messages need to be kept in queue because of network latency. Imagine network connection dropped when you are trying to send a message. In this case, the message is still in the queue, awaiting the receiver to consume upon the connection becomes available.
- Online help desk applications: Imagine a company has 5 persons working as customer support staffs. They chat with clients through a help desk application. They can talk with maximum 5 clients at a time, so other clients will be queued up. When a staff finishes serving a client, the next client in the queue is served next.
- Real-time processing applications such as screen recorder. The logic behind this kind of application is there are two threads working concurrently: The producer thread captures screenshots constantly and puts the images into a queue; the consumer thread takes the images from the queue to process the video.
- And much more.
Above we name only few types of application in which we need to use queue. Remember using queue when you need to implement producer-consumer processing.In Java, using a BlockingQueue implementation is a good choice, as its put(e) method let the producer thread waits for space to become available in the queue, and its take() method let the consumer thread waits for an element become available in the queue.1. Producer - Consumer Pseudo Code
The following is a pseudo-code of a typical producer-consumer application:class Producer implements Runnable {
private final BlockingQueue queue;
Producer(BlockingQueue q) { queue = q; }
public void run() {
while (true) { queue.put(produce()); }
}
Object produce() { ... }
}
class Consumer implements Runnable {
private final BlockingQueue queue;
Consumer(BlockingQueue q) { queue = q; }
public void run() {
while (true) { consume(queue.take()); }
}
void consume(Object x) { ... }
}
class Program {
void main() {
BlockingQueue q = new SomeBlockingQueueImplementation();
Producer p = new Producer(q);
Consumer c1 = new Consumer(q);
Consumer c2 = new Consumer(q);
new Thread(p).start();
new Thread(c1).start();
new Thread(c2).start();
}
}Here, the Producer class is a thread which constantly produces objects and put them into the queue. In practice, we should specify condition to exit the loop, such as closing/shutdown the program or a maximum number of objects reached.The Consumer class is another thread which constantly takes objects from the queue to process. In practice, we should specify condition to stop this thread by checking the queue for a special object (null, false or special value), for example:while (true) {
Integer number = queue.take();
if (number == -1) {
break;
}
consume(number);
}And in this case, the producer is responsible to put this special object into the queue to indicate there are no more elements to process.Let’s look at real examples.2. One Producer - One Consumer Example
This is a program that creates one producer thread and one consumer thread:* Producer:
import java.util.*;
import java.util.concurrent.*;
/**
* Producer using BlockingQueue example
* @author www.codejava.net
*
*/
public class Producer implements Runnable {
private BlockingQueue<Integer> queue;
public Producer (BlockingQueue<Integer> queue) {
this.queue = queue;
}
public void run() {
try {
for (int i = 0; i < 10; i++) {
queue.put(produce());
Thread.sleep(500);
}
queue.put(-1); // indicates end of producing
System.out.println("Producer STOPPED.");
} catch (InterruptedException ie) {
ie.printStackTrace();
}
}
private Integer produce() {
Integer number = new Integer((int) (Math.random() * 100));
System.out.println("Producing number => " + number);
return number;
}
} * Consumer:
import java.util.*;
import java.util.concurrent.*;
/**
* Consumer using BlockingQueue example
* @author www.codejava.net
*
*/
public class Consumer implements Runnable {
private BlockingQueue<Integer> queue;
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
public void run() {
try {
while (true) {
Integer number = queue.take();
if (number == null || number == -1) {
break;
}
consume(number);
Thread.sleep(1000);
}
System.out.println("Consumer STOPPED.");
} catch (InterruptedException ie) {
ie.printStackTrace();
}
}
private void consume(Integer number) {
System.out.println("Consuming number <= " + number);
}
} * Test Program:import java.util.*;
import java.util.concurrent.*;
/**
* Producer-Consumer test example (1 producer thread - 1 consumer thread)
* @author www.codejava.net
*
*/
public class ProducerConsumerTest {
public static void main(String[] args) {
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(20);
Thread producer = new Thread(new Producer(queue));
Thread consumer = new Thread(new Consumer(queue));
producer.start();
consumer.start();
}
}Compile and run this program would print the following output:Producing number => 21 Consuming number <= 21 Producing number => 90 Producing number => 51 Consuming number <= 90 Producing number => 23 Producing number => 61 Consuming number <= 51 Producing number => 63 Producing number => 75 Consuming number <= 23 Producing number => 99 Consuming number <= 61 Producing number => 59 Producing number => 31 Producer STOPPED. Consuming number <= 63 Consuming number <= 75 Consuming number <= 99 Consuming number <= 59 Consuming number <= 31 Consumer STOPPED.
3. One Producer - Multiple Consumers Example
In case there are multiple consumer threads, we should use the poll(time, unit) method to take elements from the head of the queue. This method does not wait forever like the take() method. Instead, it just waits for a specified of time. The lock is released either when an element found or the time out period expires. This avoids deadlock among different consumer threads.* Producer: same code as above.* Consumer:We add some code to help identify which thread is running in the output, and replace the take() method by the poll(time, unit) method. Here’s the code:import java.util.*;
import java.util.concurrent.*;
/**
* Consumer using BlockingQueue example
* @author www.codejava.net
*
*/
public class Consumer implements Runnable {
private BlockingQueue<Integer> queue;
private String threadId;
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
public void run() {
threadId = "Consumer-" + Thread.currentThread().getId();
try {
while (true) {
Integer number = queue.poll(5, TimeUnit.SECONDS);
if (number == null || number == -1) {
break;
}
consume(number);
Thread.sleep(1000);
}
System.out.println(threadId + " STOPPED.");
} catch (InterruptedException ie) {
ie.printStackTrace();
}
}
private void consume(Integer number) {
System.out.println(threadId + ": Consuming number <= " + number);
}
}Here, the consumer thread can wait maximum 5 seconds for element become available in the queue. * Producer-Consumer Test Program:
import java.util.*;
import java.util.concurrent.*;
/**
* Producer-Consumer test example (1 producer thread - N consumer threads)
* @author www.codejava.net
*
*/
public class ProducerConsumerTest {
public static void main(String[] args) {
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(20);
Thread producer = new Thread(new Producer(queue));
Thread consumer1 = new Thread(new Consumer(queue));
Thread consumer2 = new Thread(new Consumer(queue));
Thread consumer3 = new Thread(new Consumer(queue));
producer.start();
consumer1.start();
consumer2.start();
consumer3.start();
}
}Here, we create 3 consumer threads. Let run this program, it would print the following output:Producing number => 34 Consumer-11: Consuming number <= 34 Producing number => 79 Consumer-12: Consuming number <= 79 Producing number => 43 Consumer-13: Consuming number <= 43 Producing number => 44 Consumer-11: Consuming number <= 44 Producing number => 49 Consumer-12: Consuming number <= 49 Producing number => 28 Consumer-13: Consuming number <= 28 Producing number => 2 Consumer-11: Consuming number <= 2 Producing number => 92 Consumer-12: Consuming number <= 92 Producing number => 89 Consumer-13: Consuming number <= 89 Producing number => 85 Consumer-11: Consuming number <= 85 Producer STOPPED. Consumer-12 STOPPED. Consumer-13 STOPPED. Consumer-11 STOPPED.We hope these examples give you the ideas and help you implement similar scenarios in your program. Feel free to help us improve the code by replying in the comments section.
Related Queue tutorials:
Other Java Collections Tutorials:
- Java Set Collection Tutorial and Examples
- Java Map Collection Tutorial and Examples
- Java List Collection Tutorial and Examples
- Understanding equals() and hashCode() in Java
- Understanding Object Ordering in Java with Comparable and Comparator
- 18 Java Collections and Generics Best Practices
About the Author:
Nam Ha Minh is certified Java programmer (SCJP and SCWCD). He started programming with Java in the time of Java 1.4 and has been falling in love with Java since then. Make friend with him on Facebook and watch his Java videos you YouTube.
Comments
Thanks.