Java BlockingQueue examples
In Java, we can use BlockingQueue
to create a queue which shared by both producer and the consumer.
- Producer – Generate data and put it into the queue.
- Consumer – Remove the data from the queue.
Read this article to understand what is producer and consumer.
The BlockingQueue implementations are thread-safe, safely be used with multiple producers and multiple consumers.
1. BlockingQueue
A simple BlockingQueue
example, a producer generates data and put it into a queue, at the same time, a consumer takes the data from the same queue.
1.1 Producer – A Runnable
object to put 20 integers into a queue.
package com.mkyong.concurrency.queue.simple.raw;
import java.util.concurrent.BlockingQueue;
public class Producer implements Runnable {
private final BlockingQueue<Integer> queue;
@Override
public void run() {
try {
process();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void process() throws InterruptedException {
// Put 20 ints into Queue
for (int i = 0; i < 20; i++) {
System.out.println("[Producer] Put : " + i);
queue.put(i);
System.out.println("[Producer] Queue remainingCapacity : " + queue.remainingCapacity());
Thread.sleep(100);
}
}
public Producer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
}
1.2 Consumer – A Runnable
object to take items from a queue.
package com.mkyong.concurrency.queue.simple.raw;
import java.util.concurrent.BlockingQueue;
public class Consumer implements Runnable {
private final BlockingQueue<Integer> queue;
@Override
public void run() {
try {
while (true) {
Integer take = queue.take();
process(take);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void process(Integer take) throws InterruptedException {
System.out.println("[Consumer] Take : " + take);
Thread.sleep(500);
}
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
}
1.3 Run it. Start 1 producer and 1 consumer, and create a queue with size of 10.
package com.mkyong.concurrency.queue.simple;
import com.mkyong.concurrency.queue.simple.raw.Consumer;
import com.mkyong.concurrency.queue.simple.raw.Producer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class Main {
public static void main(String[] args) {
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
new Thread(new Producer(queue)).start();
new Thread(new Consumer(queue)).start();
}
}
Output
The producer won’t try to put more data into the queue if it is full.
[Producer] Put : 0
[Producer] Queue remainingCapacity : 9
[Consumer] Take : 0
[Producer] Put : 1
[Producer] Queue remainingCapacity : 9
[Producer] Put : 2
[Producer] Queue remainingCapacity : 8
[Producer] Put : 3
[Producer] Queue remainingCapacity : 7
[Producer] Put : 4
[Producer] Queue remainingCapacity : 6
[Producer] Put : 5
[Producer] Queue remainingCapacity : 5
[Consumer] Take : 1
[Producer] Put : 6
[Producer] Queue remainingCapacity : 5
[Producer] Put : 7
[Producer] Queue remainingCapacity : 4
[Producer] Put : 8
[Producer] Queue remainingCapacity : 3
[Producer] Put : 9
[Producer] Queue remainingCapacity : 2
[Producer] Put : 10
[Producer] Queue remainingCapacity : 1
[Consumer] Take : 2
[Producer] Put : 11
[Producer] Queue remainingCapacity : 1
[Producer] Put : 12
[Producer] Queue remainingCapacity : 0
[Producer] Put : 13
[Consumer] Take : 3
[Producer] Queue remainingCapacity : 0
[Producer] Put : 14
[Consumer] Take : 4
[Producer] Queue remainingCapacity : 0
[Producer] Put : 15
[Consumer] Take : 5
[Producer] Queue remainingCapacity : 0
[Producer] Put : 16
[Consumer] Take : 6
[Producer] Queue remainingCapacity : 0
[Producer] Put : 17
[Consumer] Take : 7
[Producer] Queue remainingCapacity : 0
[Producer] Put : 18
[Consumer] Take : 8
[Producer] Queue remainingCapacity : 0
[Producer] Put : 19
[Consumer] Take : 9
[Producer] Queue remainingCapacity : 0
[Consumer] Take : 10
[Consumer] Take : 11
[Consumer] Take : 12
[Consumer] Take : 13
[Consumer] Take : 14
[Consumer] Take : 15
[Consumer] Take : 16
[Consumer] Take : 17
[Consumer] Take : 18
[Consumer] Take : 19
The program will not stop or exit, it will keep on running there to put and take data from the BlockingQueue
2. BlockingQueue + Poison Pill
The “poison pill” is a general solution to stop or interrupt both producer and consumer threads. The idea is the producer put a “poison pill” into the queue and exit, if the “consumer” see the “poison pill” then stop and exit.
2.1 A producer with poison pill solution.
package com.mkyong.concurrency.queue.simple.poison;
import java.util.concurrent.BlockingQueue;
public class ProducerPoison implements Runnable {
private final BlockingQueue<Integer> queue;
private final Integer POISON;
@Override
public void run() {
try {
process();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
while (true) {
try {
queue.put(POISON);
break;
} catch (InterruptedException e) {
//...
}
}
}
}
private void process() throws InterruptedException {
// Put 20 elements into Queue
for (int i = 0; i < 20; i++) {
System.out.println("[Producer] Put : " + i);
queue.put(i);
System.out.println("[Producer] Queue remainingCapacity : " + queue.remainingCapacity());
Thread.sleep(100);
}
}
public ProducerPoison(BlockingQueue<Integer> queue, Integer POISON) {
this.queue = queue;
this.POISON = POISON;
}
}
2.2 A consumer with poison pill solution.
package com.mkyong.concurrency.queue.simple.poison;
import java.util.concurrent.BlockingQueue;
public class ConsumerPoison implements Runnable {
private final BlockingQueue<Integer> queue;
private final Integer POISON;
@Override
public void run() {
try {
while (true) {
Integer take = queue.take();
process(take);
// if this is a poison pill, break, exit
if (take == POISON) {
break;
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void process(Integer take) throws InterruptedException {
System.out.println("[Consumer] Take : " + take);
Thread.sleep(500);
}
public ConsumerPoison(BlockingQueue<Integer> queue, Integer POISON) {
this.queue = queue;
this.POISON = POISON;
}
}
2.3 Start 2 producers and 2 consumers.
package com.mkyong.concurrency.queue.simple;
import com.mkyong.concurrency.queue.simple.poison.ConsumerPoison;
import com.mkyong.concurrency.queue.simple.poison.ProducerPoison;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class Main {
public static void main(String[] args) {
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
//new Thread(new Producer(queue)).start();
//new Thread(new Consumer(queue)).start();
Integer poison = -1;
new Thread(new ProducerPoison(queue, poison)).start();
new Thread(new ProducerPoison(queue, poison)).start();
new Thread(new ConsumerPoison(queue, poison)).start();
new Thread(new ConsumerPoison(queue, poison)).start();
}
}
Output
[Producer] Put : 0
[Producer] Put : 0
[//...
[Consumer] Take : 18
[Consumer] Take : 18
[Consumer] Take : 19
[Consumer] Take : 19
[Consumer] Take : -1
[Consumer] Take : -1
Process finished with exit code 0
3. BlockingQueue + File Indexing
A BlockingQueue
example to create a simple file indexing engine. A producer crawls a directory and puts the filename into the queue, at the same time, the consumer take the filename from the same queue and index it.
3.1 Producer.
package com.mkyong.concurrency.queue.crawler;
import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.BlockingQueue;
// Producer
// Crawl file system and put the filename in BlockingQueue.
public class FileCrawlerProducer implements Runnable {
private final BlockingQueue<File> fileQueue;
private final FileFilter fileFilter;
private final File file;
private final File POISON;
private final int N_POISON_PILL_PER_PRODUCER;
@Override
public void run() {
try {
crawl(file);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
while (true) {
try {
System.out.println(Thread.currentThread().getName()
+ " - FileCrawlerProducer is done, try poison all the consumers!");
// poison all threads
for (int i = 0; i < N_POISON_PILL_PER_PRODUCER; i++) {
System.out.println(Thread.currentThread().getName() + " - puts poison pill!");
fileQueue.put(POISON);
}
break;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public FileCrawlerProducer(BlockingQueue<File> fileQueue,
FileFilter fileFilter, File file,
File POISON, int n_POISON_PILL_PER_PRODUCER) {
this.fileQueue = fileQueue;
this.fileFilter = fileFilter;
this.file = file;
this.POISON = POISON;
N_POISON_PILL_PER_PRODUCER = n_POISON_PILL_PER_PRODUCER;
}
private void crawl(File root) throws InterruptedException {
File[] entries = root.listFiles(fileFilter);
if (entries != null) {
for (File entry : entries) {
if (entry.isDirectory()) {
crawl(entry);
} else if (!isIndexed(entry)) {
System.out.println("[FileCrawlerProducer] - Found..."
+ entry.getAbsoluteFile());
fileQueue.put(entry);
}
}
}
}
private boolean isIndexed(File f) {
return false;
}
}
3.2 Consumer.
package com.mkyong.concurrency.queue.crawler;
import java.io.File;
import java.util.concurrent.BlockingQueue;
// Consumer
public class IndexerConsumer implements Runnable {
private final BlockingQueue<File> fileQueue;
private final File POISON;
@Override
public void run() {
try {
while (true) {
File take = fileQueue.take();
if (take == POISON) {
System.out.println(Thread.currentThread().getName() + " die");
break;
}
indexFile(take);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public void indexFile(File file) {
if (file.isFile()) {
System.out.println(Thread.currentThread().getName()
+ " [IndexerConsumer] - Indexing..." + file.getAbsoluteFile());
}
}
public IndexerConsumer(BlockingQueue<File> fileQueue, File POISON) {
this.fileQueue = fileQueue;
this.POISON = POISON;
}
}
3.3 Start 1 producer and 2 consumers.
package com.mkyong.concurrency.queue.crawler;
import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class Main {
private static final File POISON = new File("This is a POISON PILL");
public static void main(String[] args) {
int N_PRODUCERS = 1;
int N_CONSUMERS = 2;//Runtime.getRuntime().availableProcessors();
int N_POISON_PILL_PER_PRODUCER = N_CONSUMERS / N_PRODUCERS;
int N_POISON_PILL_REMAIN = N_CONSUMERS % N_PRODUCERS;
System.out.println("N_PRODUCERS : " + N_PRODUCERS);
System.out.println("N_CONSUMERS : " + N_CONSUMERS);
System.out.println("N_POISON_PILL_PER_PRODUCER : " + N_POISON_PILL_PER_PRODUCER);
System.out.println("N_POISON_PILL_REMAIN : " + N_POISON_PILL_REMAIN);
//unbound queue, no limit
BlockingQueue<File> queue = new LinkedBlockingQueue<>();
FileFilter filter = new FileFilter() {
public boolean accept(File file) {
return true;
}
};
File root = new File("C:\\users");
for (int i = 0; i < N_PRODUCERS - 1; i++) {
new Thread(new FileCrawlerProducer(queue, filter, root,
POISON, N_POISON_PILL_PER_PRODUCER)).start();
}
new Thread(new FileCrawlerProducer(queue, filter, root, POISON,
N_POISON_PILL_PER_PRODUCER + N_POISON_PILL_REMAIN)).start();
for (int i = 0; i < N_CONSUMERS; i++) {
new Thread(new IndexerConsumer(queue, POISON)).start();
}
}
}
Output
//...
[FileCrawlerProducer] - Found...C:\users\Public\Videos\desktop.ini
Thread-2 [IndexerConsumer] - Indexing...C:\users\Public\Videos\desktop.ini
Thread-0 - FileCrawlerProducer is done, try poison all the consumers!
Thread-0 - puts poison pill!
Thread-0 - puts poison pill!
Thread-1 die
Thread-2 die
Process finished with exit code 0
Hello and thank you for the great tutorials
One question: is the “.start()” correct since the IndexerConsumer implements “run()” function. The same applies for every example.
for (int i = 0; i < N_CONSUMERS; i++) {
new Thread(new IndexerConsumer(queue, POISON)).start();
}