Как предотвратить голодание сообщений с низким приоритетом в приоритетной очереди ActiveMQ?

Я работаю над системой, в которой нам нужно реализовать приоритетную очередь. У нас есть сообщения с разными приоритетами, и нам нужно обрабатывать сообщения на основе приоритета. Прямо сейчас мы планируем использовать ActiveMQ в качестве нашей технологии организации очередей по многим причинам, одна из которых заключается в том, что она поддерживает очереди с приоритетом.

Что лучше всего делать с очередью с приоритетами в ActiveMQ? Чтобы быть точным, нам нужно гарантировать, что даже сообщение с низким приоритетом в конечном итоге будет обработано, даже если сообщения с более высоким приоритетом продолжают наводнять очередь. Есть ли в ActiveMQ что-то встроенное? Или нам нужно создать что-то свое, чтобы повышать приоритет по мере устаревания сообщения?


java queue message-queue activemq priority-queue
person Matthew Kubicina    schedule 18.06.2011    source источник
comment
Просто подумав об этом, не возможно ли, что поток низкоприоритетных сообщений, если в конечном итоге увеличится в приоритете, истощит законно высокоприоритетные сообщения? и если сообщения с высоким приоритетом приходят так быстро, не пора ли выключить систему для ремонта?   -  person jcomeau_ictx    schedule 18.06.2011

Ответы (1)


arrow_upward
4
arrow_downward

основной способ сделать это — повысить приоритет, когда сообщение становится старше.

Таким образом, сообщение с низким приоритетом, скажем, час назад, имеет более высокий приоритет, чем новое сообщение с высоким приоритетом.

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;
    }

}

как отмечалось в комментариях, поток сообщений с низким приоритетом, отправленных несколько часов назад, временно приведет к голоданию новых сообщений с высоким приоритетом, и для их правильного распределения потребуется более сложный метод.

другой метод использует несколько очередей, по одной для каждого приоритета, и извлекает несколько из очереди с более высоким приоритетом для каждой очереди из очереди с низким приоритетом.

этот последний метод действительно жизнеспособен только для небольшого количества приоритетов, в то время как первый метод, который я предоставил, может обрабатывать произвольное количество приоритетов.

person ratchet freak    schedule 18.06.2011