How to Remove Elements from PriorityQueue according to a Element Property?

Question

I am try to set Maximum Waiting Time inside my PQueue. This Maximum Waiting Time will check my PQueue automatically if there are any links are waiting more than Maximum Waiting Time to remove it. I did this changes to my code it is working but it is stopping exactly after removing the links. I want to remove all elements from my PQueue according to waiting time condition. Can you tell me what I am missing here ?

This is my class:

public class MyClass {

    public static PriorityQueue <LinkNodeLight> PQueue = new PriorityQueue <> (); 


    private static Set<String> DuplicationLinksHub = new LinkedHashSet <> ();         

    private static Integer IntraLinkCount = new Integer (0);                 
    private static Integer InterLinkCount = new Integer (0);                 
    private static Integer DuplicationLinksCount = new Integer (0);     
    private static Integer MaxWaitTime = new Integer (60000); // 1 M= 60000 MS


    @SuppressWarnings("null")
    LinkNode deque(){

        LinkNode link = null;
        synchronized (PQueue) {

            link = (LinkNode) PQueue.poll();
            if (link != null) {
                link.setDequeTime(new DateTime());
                if (link.isInterLinks())
                    synchronized (InterLinkCount) {
                        InterLinkCount--;
                        }
                else
                    synchronized (IntraLinkCount) {
                        IntraLinkCount--;
                        }
            }

            synchronized (PQueue) {
                if (link.waitingInQueue()>MaxWaitTime) {

                    link = (LinkNode) PQueue.remove();
                                    System.out.println("*********************************");
                                    System.out.println("This Link is Deopped: " + link);
                                    System.out.println("%%% MaX Waiting Time:" + (MaxWaitTime/60000)+"Min");

                                    System.out.println("*********************************");
                  }
            }
            return link;


        }

Answer 1

Your question is a bit opaque, but if I understand it correctly you want to check your PriorityQueue to see if there are items that have waited longer than a specific time.

Your usage of synchronized on the IntraLinkCount and InterLinkCount is, as has been mentioned already, a bit strange. There is a fairly unknown alternative, the atomic integer class AtomicInteger (in the package java.util.concurrent.atomic:

private static AtomicInteger IntraLinkCount = Integer.valueOf(0);

This will work as you want.

The second problem is that you use the poll() method. This will remove the top item from the queue. Maybe you want to use peek() instead, and then only use remove() if the returned link object satisfies link.waitingInQueue() > MaxWaitTime?

By the way, your queue will return items according to their "natural ordering". This means the compareTo method is used, and the "smallest" will be returned from the queue first. I think you might want to implement a custom compareTo that puts the longest waiting link first instead?

You could also create your PriorityQueue with a custom Comparator object instead.

Something like this:

public class MyClass {
    public static PriorityQueue<LinkNodeLight> PQueue = new PriorityQueue<>();

    private static AtomicInteger IntraLinkCount = new AtomicInteger(0);
    private static AtomicInteger InterLinkCount = new AtomicInteger(0);

    private static Integer MaxWaitTime = Integer.valueOf(60_000); // 1 M= 60000 MS

    LinkNode deque() {
        LinkNode link = null;

        synchronized (PQueue) {
            link = PQueue.peek();

            if (link != null) {
                link.setDequeTime(LocalDateTime.now());

                if (link.isInterLinks())
                    InterLinkCount.decrementAndGet();
                else
                    IntraLinkCount.decrementAndGet();

                if (link.waitingInQueue() > MaxWaitTime) {
                    link = PQueue.remove();

                    System.out.println("*********************************");
                    System.out.println("This Link is Deopped: " + link);
                    System.out.println("%%% MaX Waiting Time:" + MaxWaitTime / 60000 + "Min");
                    System.out.println("*********************************");

                    return link;
                } else
                    return null;
            }
        }

        return link; // Not sure what you want to return here
    }
}

If you are lucky enough to be on Java 8, some magic like this might be useful instead:

synchronized (PQueue) {
    link = PQueue.stream().filter(node -> node.waitingInQueue() > MaxWaitTime).findFirst().orElse(null);

    if (link != null)
        PQueue.remove(link);
}