Bael 845 transfer queue (#1697)

* BAEL-845 tranferqueue article

* BAEL-845 m to m example

* BAEL-845 move code to test

* BAEL-845 use tryTransfer

* BAEL-845 proper if logic

* BAEL-845 proper test

* BAEL-845 robust test
This commit is contained in:
Tomasz Lelek
2017-04-20 23:05:44 +02:00
committed by pedja4
parent 76673a33f1
commit 2648a4ec25
3 changed files with 21 additions and 3 deletions

View File

@@ -1,11 +1,13 @@
package com.baeldung.transferqueue;
import java.util.concurrent.TransferQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class Consumer implements Runnable {
private final TransferQueue<String> transferQueue;
private final String name;
private final int numberOfMessagesToConsume;
public final AtomicInteger numberOfConsumedMessages = new AtomicInteger();
public Consumer(TransferQueue<String> transferQueue, String name, int numberOfMessagesToConsume) {
this.transferQueue = transferQueue;
@@ -28,6 +30,7 @@ public class Consumer implements Runnable {
}
private void longProcessing(String element) throws InterruptedException {
Thread.sleep(1_000);
numberOfConsumedMessages.incrementAndGet();
Thread.sleep(500);
}
}

View File

@@ -2,11 +2,13 @@ package com.baeldung.transferqueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TransferQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class Producer implements Runnable {
private final TransferQueue<String> transferQueue;
private final String name;
private final Integer numberOfMessagesToProduce;
public final AtomicInteger numberOfProducedMessages = new AtomicInteger();
public Producer(TransferQueue<String> transferQueue, String name, Integer numberOfMessagesToProduce) {
this.transferQueue = transferQueue;
@@ -23,6 +25,7 @@ public class Producer implements Runnable {
if (!added) {
System.out.println("can not add an element due to the timeout");
} else {
numberOfProducedMessages.incrementAndGet();
System.out.println("Producer: " + name + " transferred element: A" + i);
}
} catch (InterruptedException e) {