I was coding for the following programming question:
Suppose you have two independent blocking queues that will keep getting new data, the new data will always be greater than the last element of the queue it is going to. You can only use getNext() to get the data from these two blocking queues and each data can be fetched only once. Write a program to output all pairs of data from these two blocking queues that have difference smaller than 1.
My implementation uses three additional queues with two holding data from the two given blocking queues, and one holding the result. I used two threads each handling one of the queue, and send the output result to the result queue. Whenever a new data comes into the first queue, I will pop every data in the second queue that has difference larger or equal to 1, then looping through second queue to output all pairs smaller than 1.
Here is my implementation:
The BlockingQueueCalculator class handles a individual queue and performs pair calculation. It will wait for new element from the input Blocking Queue, and will terminate whenever fed a POISON. The lock object will prevent race condition, so that only one queue is allowed to process at a time.
public class BlockingQueueCalculator implements Runnable {
public static final double POISON = Double.NEGATIVE_INFINITY;
private final Deque<Double> q1;
private final Deque<Double> q2;
private final BlockingDeque<Double> inputQueue;
private final Deque<Tuple<Double, Double>> pairsQueue;
private final Object lock;
static class Tuple<T, V> {
T t;
V v;
public Tuple(T t, V v) {
this.t = t;
this.v = v;
}
}
public BlockingQueueCalculator(Deque<Double> q1,
Deque<Double> q2,
Object lock,
BlockingDeque<Double> input,
Deque<Tuple<Double, Double>> pairs) {
this.q1 = q1;
this.q2 = q2;
this.lock = lock;
this.inputQueue = input;
this.pairsQueue = pairs;
}
public void calculate(double timestamp) throws InterruptedException {
if (!q2.isEmpty()) {
while (!q2.isEmpty() && Math.abs(q2.peekFirst() - timestamp) >= 1.0) {
System.out.println("Discarding from q2: " + q2.pollFirst());
}
for (Double oldTimestamp : q2) {
System.out.println("Checking q2 element: " + oldTimestamp);
if (oldTimestamp == null || Math.abs(oldTimestamp - timestamp) >= 1.0) {
return;
}
System.out.println("Adding to Pairs: " + oldTimestamp + ", " + timestamp);
pairsQueue.add(new Tuple<>(oldTimestamp, timestamp));
}
}
}
@Override
public void run() {
try {
double timeStamp;
synchronized (lock) {
while ((timeStamp = inputQueue.takeFirst()) != POISON) {
System.out.println("Adding to q1: " + timeStamp);
q1.add(timeStamp);
calculate(timeStamp);
}
}
} catch (InterruptedException e) {
System.out.println("Thread interrupted: " + e);
Thread.currentThread().interrupt();
}
}
}
The BlockingQueuePairFinder is the main thread that setup environment and start other worker threads.
public class BlockingQueuePairFinder {
private final BlockingDeque<Double> inputQueue1;
private final BlockingDeque<Double> inputQueue2;
private final Deque<Double> q1 = new LinkedList<>();
private final Deque<Double> q2 = new LinkedList<>();
private final Deque<BlockingQueueCalculator.Tuple<Double, Double>> pairsList =
new LinkedList<>();
private final Object lock = new Object();
private Thread t1;
private Thread t2;
public BlockingQueuePairFinder(BlockingDeque<Double> inputQueue1,
BlockingDeque<Double> inputQueue2) {
this.inputQueue1 = inputQueue1;
this.inputQueue2 = inputQueue2;
this.t1 = new Thread(new BlockingQueueCalculator(q1, q2, lock, inputQueue1, pairsList));
this.t2 = new Thread(new BlockingQueueCalculator(q2, q1, lock, inputQueue2, pairsList));
}
public void run() {
if(t1 == null || !t1.isAlive()) {
t1 = new Thread(new BlockingQueueCalculator(q1, q2, lock, inputQueue1, pairsList));
t1.start();
}
if(t2 == null || !t2.isAlive()) {
t2 = new Thread(new BlockingQueueCalculator(q2, q1, lock, inputQueue2, pairsList));
t2.start();
}
}
public static void main(String...args) {
BlockingDeque<Double> q1 = new LinkedBlockingDeque<>();
BlockingDeque<Double> q2 = new LinkedBlockingDeque<>();
BlockingQueuePairFinder finder = new BlockingQueuePairFinder(q1, q2);
finder.run();
q1.offerLast(0.2);
q2.offerLast(1.0);
q1.offerLast(1.4);
q1.offerLast(3.0);
q2.offerLast(1.1);
q2.offerLast(3.5);
q1.offerLast(BlockingQueueCalculator.POISON);
q2.offerLast(BlockingQueueCalculator.POISON);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
for(BlockingQueueCalculator.Tuple<Double, Double> tuple : finder.pairsList) {
System.out.println(tuple.t + ", " + tuple.v);
}
}
}
Is this a good OO and concurrency design? If not, what design would you propose?
Also, given the fact that I used lock(semaphore?) to allow only one queue to operate at any time, if I generalize this design to accept a List<BlockingDeque>, then my design would simply block all but one queue at any time. Is this an optimal solution and what would you change under such circumstance?
