2
\$\begingroup\$

A followup on my previous question Using blocking queue to calculate data stream. This time suppose we have a list of blocking queues as stated in the following manner:

Suppose you have a list of 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 blocking queues and each data can be fetched only once. Write a program to output all pairs of data from these blocking queues that have difference smaller than 1.

The following is my implementation:

GeneralizedBlockingQueueCalculator will still have only one thread running at any given time to process the new data and calculate pairs. But this time, we can only remove data from queue if the smallest element of all latest data from each blocking queue is larger than head of current queue by 1. Therefore I used a minimum Heap to hold references to other queues.

public class GeneralizedBlockingQueueCalculator implements Runnable {
    public static final double POISON = Double.NEGATIVE_INFINITY;
    private final Deque<Double> q1;
    private final Queue<Deque<Double>> minHeap;
    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 GeneralizedBlockingQueueCalculator(Deque<Double> q1,
                                              Queue<Deque<Double>> minHeap,
                                              Object lock,
                                              BlockingDeque<Double> input,
                                              Deque<Tuple<Double, Double>> pairs) {
        this.q1 = q1;
        this.minHeap = minHeap;
        this.lock = lock;
        this.inputQueue = input;
        this.pairsQueue = pairs;
    }

    public void calculate(double timestamp) {
        if (minHeap == null || minHeap.isEmpty()) {
            return;
        }
        Deque<Double> topDeque = minHeap.peek();
        if (topDeque == null) {
            return;
        }
        Double minNewElement = topDeque.peekLast();

        Iterator<Deque<Double>> iterator = minHeap.iterator();
        while (iterator.hasNext()) {
            Deque<Double> curDeque = iterator.next();
            if (curDeque == null || curDeque.isEmpty() || curDeque == q1) {
                continue;
            }

            Iterator<Double> dequeIterator = curDeque.iterator();
            while (dequeIterator.hasNext()) {
                Double curDouble = dequeIterator.next();
                System.out.println("Checking: " + curDouble);
                if (minNewElement != null && minNewElement - curDouble >= 1.0) {
                    System.out.println("Removing: " + curDouble);
                    dequeIterator.remove();
                } else if (Math.abs(timestamp - curDouble) < 1.0) {
                    System.out.println("Adding pair: " + curDouble + ", " + timestamp);
                    pairsQueue.add(new Tuple<>(curDouble, timestamp));
                }
            }
        }
    }

    @Override
    public void run() {
        try {
            double timeStamp;
            synchronized (lock) {
                while ((timeStamp = inputQueue.takeFirst()) != POISON) {
                    minHeap.remove(q1);
                    q1.add(timeStamp);
                    minHeap.add(q1);
                    System.out.println("Adding: " + timeStamp);
                    calculate(timeStamp);
                }
            }
        } catch (InterruptedException e) {
            System.out.println("Thread interrupted: " + e);
            Thread.currentThread().interrupt();
        }
    }
}

The GeneralizedBlockingQueuePairFinder is almost identical as before, besides an added fixed thread pool, and constructing minimum heap.

public class GeneralizedBlockingQueuePairFinder {
    private final ExecutorService threadPool;
    private final Queue<Deque<Double>> minHeap;
    private final List<BlockingDeque<Double>> blockingDequeList;
    private final Deque<GeneralizedBlockingQueueCalculator.Tuple<Double, Double>> pairsList =
            new LinkedList<>();
    private final Object lock = new Object();

    public GeneralizedBlockingQueuePairFinder(List<BlockingDeque<Double>> blockingDequeList) {
        threadPool = Executors.newFixedThreadPool(blockingDequeList.size());
        minHeap = new PriorityQueue<>(blockingDequeList.size(), (o1, o2) -> {
            Double d1 = o1.peekLast();
            Double d2 = o2.peekLast();
            if (d1 == null) {
                return -1;
            }
            if (d2 == null) {
                return 1;
            }

            return d1.compareTo(d2);
        });
        for (int i = 0; i < blockingDequeList.size(); i++) {
            minHeap.add(new LinkedList<>());
        }
        this.blockingDequeList = blockingDequeList;
    }

    public void run() {
        int index = 0;
        Iterator<Deque<Double>> dequeIterator = minHeap.iterator();
        while (dequeIterator.hasNext()) {
            Deque<Double> deque = dequeIterator.next();
            BlockingDeque<Double> blockingDeque = blockingDequeList.get(index);
            threadPool.submit(new GeneralizedBlockingQueueCalculator(deque, minHeap, lock,
                    blockingDeque, pairsList));
            index++;
        }
        threadPool.shutdown();
    }

    public static void main(String... args) {
        List<BlockingDeque<Double>> list = new ArrayList<>();
        BlockingDeque<Double> q1 = new LinkedBlockingDeque<>();
        BlockingDeque<Double> q2 = new LinkedBlockingDeque<>();
        BlockingDeque<Double> q3 = new LinkedBlockingDeque<>();
        list.add(q1);
        list.add(q2);
        list.add(q3);

        GeneralizedBlockingQueuePairFinder finder = new GeneralizedBlockingQueuePairFinder(list);
        finder.run();

        q1.offerLast(0.1);
        q2.offerLast(0.2);
        q1.offerLast(3.0);
        q3.offerLast(0.3);
        q2.offerLast(0.4);
        q3.offerLast(5.0);
        q1.offerLast(5.0);
        q2.offerLast(0.6);
        q2.offerLast(1.2);
        q3.offerLast(10.2);
        q1.offerLast(11.0);
        q3.offerLast(20.3);

        q1.offerLast(GeneralizedBlockingQueueCalculator.POISON);
        q2.offerLast(GeneralizedBlockingQueueCalculator.POISON);
        q3.offerLast(GeneralizedBlockingQueueCalculator.POISON);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {

        }

        for (GeneralizedBlockingQueueCalculator.Tuple<Double, Double> tuple : finder.pairsList) {
            System.out.println(tuple.t + ", " + tuple.v);
        }
    }
}

Again, any OO and concurrency design improvements and suggestions are welcome.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.