iSharkFly-Docs
/
java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

BAEL-614 introduced poison pill message to stop our producers form run indefinitely

This commit is contained in:
Tomasz Lelek 2017-01-28 10:45:57 +01:00
parent c9a1fb1b39
commit 2c0494dd03
3 changed files with 18 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
core-java/src/main/java/com/baeldung/concurrent/blockingqueue/BlockingQueueUsage.java
View File

@ -9,16 +9,18 @@ public class BlockingQueueUsage {
int BOUND = 10; int BOUND = 10;
int N_PRODUCERS = 4; int N_PRODUCERS = 4;
int N_CONSUMERS = Runtime.getRuntime().availableProcessors(); int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
int poisonPill = Integer.MAX_VALUE;
int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS;
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(BOUND); BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(BOUND);
for (int i = 0; i < N_PRODUCERS; i++) { for (int i = 0; i < N_PRODUCERS; i++) {
new Thread(new NumbersProducer(queue)).start(); new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer)).start();
} }
for (int j = 0; j < N_CONSUMERS; j++) { for (int j = 0; j < N_CONSUMERS; j++) {
new Thread(new NumbersConsumer(queue)).start(); new Thread(new NumbersConsumer(queue, poisonPill)).start();
} }
} }
} }

7
core-java/src/main/java/com/baeldung/concurrent/blockingqueue/NumbersConsumer.java
View File

@ -5,15 +5,20 @@ import java.util.concurrent.BlockingQueue;
class NumbersConsumer implements Runnable { class NumbersConsumer implements Runnable {
private final BlockingQueue<Integer> queue; private final BlockingQueue<Integer> queue;
private final int poisonPill;
public NumbersConsumer(BlockingQueue<Integer> queue) { public NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) {
this.queue = queue; this.queue = queue;
this.poisonPill = poisonPill;
} }
public void run() { public void run() {
try { try {
while (true) { while (true) {
Integer number = queue.take(); Integer number = queue.take();
if (number.equals(poisonPill)) {
return;
}
String result = number.toString(); String result = number.toString();
System.out.println(Thread.currentThread().getName() + " result: " + result); System.out.println(Thread.currentThread().getName() + " result: " + result);

9
core-java/src/main/java/com/baeldung/concurrent/blockingqueue/NumbersProducer.java
View File

@ -8,10 +8,14 @@ import java.util.concurrent.BlockingQueue;
class NumbersProducer implements Runnable { class NumbersProducer implements Runnable {
private final Random random = new Random(); private final Random random = new Random();
private final BlockingQueue<Integer> numbersQueue; private final BlockingQueue<Integer> numbersQueue;
private final int poisonPill;
private final int poisonPillPerProducer;
public NumbersProducer(BlockingQueue<Integer> numbersQueue) { public NumbersProducer(BlockingQueue<Integer> numbersQueue, int poisonPill, int poisonPillPerProducer) {
this.numbersQueue = numbersQueue; this.numbersQueue = numbersQueue;
this.poisonPill = poisonPill;
this.poisonPillPerProducer = poisonPillPerProducer;
} }
@ -27,5 +31,8 @@ class NumbersProducer implements Runnable {
for (int i = 0; i < 100; i++) { for (int i = 0; i < 100; i++) {
numbersQueue.put(random.nextInt(100)); numbersQueue.put(random.nextInt(100));
} }
for (int j = 0; j < poisonPillPerProducer; j++) {
numbersQueue.put(poisonPill);
}
} }
} }