How to prevent starvation of low priority messages in ActiveMQ priority queue?

Question

I am working on a system where we need to implement a priority queue. We have messages with different priorities, and we need to process messages based on priority. We are currently striving to use ActiveMQ as our queuing technology for many reasons, one of which is support for priority queues.

With priority in ActiveMQ, what is the best way to fight hunger? To be specific, we must ensure that even a low priority message is ultimately processed, even if higher priority messages continue to flood the queue. Does ActiveMQ have something built in? Or do we need to build something of our own to increase priority, because the age of the messages?

Answer 1

The main way to do this is to increase priority when the message is older

thus a low priority message, say an hour ago, has a higher priority than a new high priority message

public class Message implements Comparable<Message>{ private final long time;//timestamp from creation (can be altered to insertion in queue) in millis the lower this value the older the message (and more important that it needs to be handled) private final int pr;//priority the higher the value the higher the priority /** * the offset that the priority brings currently set for 3 hours * * meaning a message with pr==1 has equal priority than a message with pr==0 from 3 hours ago */ private static final long SHIFT=3*60*60*1000; public Message(int priority){ this.pr=priority; this.time = System.currentTimeMillis(); } //I'm assuming here the priority sorting is done with natural ordering public boolean compareTo(Message other){ long th = this.time-this.pr*SHIFT; long ot = other.time-other.pr*SHIFT; if(th<ot)return 1; if(th>ot)return -1; return 0; } } 

as noted in the comments, however, the flow of messages with low priority from a few hours ago will temporarily starve for new messages with high priority, and to ensure their proper use will require a more complex method.

---

the other method uses several queues, one for each priority, and takes several from a higher priority queue for each of the low priority queued queues

this last method is really effective for a small number of priorities, while the first method I provided can handle an arbitrary number of priorities