Как избежать дублирования сообщений в JMS / ActiveMQ - PullRequest
Новая
       54

Как избежать дублирования сообщений в JMS / ActiveMQ

19 голосов
/ 08 февраля 2011

Есть ли способ подавления дублирующихся сообщений в очереди, определенной на сервере ActiveMQ?

Я пытался определить JMSMessageID вручную (message.setJMSMessageID ("uniqueid")), но сервер игнорирует эту модификацию и доставляетсообщение со встроенным сгенерированным JMSMessageID.

По спецификации я не нашел ссылки о том, как дедуплицировать сообщения.

В HornetQ, чтобы справиться с этой проблемой, нам нужно объявитьспецифическое свойство HQ org.hornetq.core.message.impl.HDR_DUPLICATE_DETECTION_ID для определения сообщения.

т.е.: 

Message jmsMessage = session.createMessage();
String myUniqueID = "This is my unique id"; // Could use a UUID for this
message.setStringProperty(HDR_DUPLICATE_DETECTION_ID.toString(), myUniqueID);

Кто-то знает, существует ли подобное решение для ActiveMQ?

Ответы [ 5 ]

7 голосов
/ 09 февраля 2011

Вам следует взглянуть на Apache Camel, он предоставляет потребительский компонент Idempotent, который будет работать с любым провайдером JMS, см .: http://camel.apache.org/idempotent-consumer.html

Использование этого в сочетании с компонентом ActiveMQ делает использование JMS довольно простым, см.: http://camel.apache.org/activemq.html

5 голосов
/ 09 февраля 2011

Я сомневаюсь, что ActiveMQ поддерживает его изначально, но это должно быть легко реализовать идемпотентного потребителя. Один из способов сделать это - добавить уникальный идентификатор к каждому сообщению на стороне производителя, а теперь на стороне потребителя, используя хранилище (БД, кэш и т. Д.), Можно проверить, было ли получено сообщение раньше, и продолжить обработку на основе этой проверки.

Я вижу предыдущий вопрос о стекопереработке в том же ключе - Apache ActiveMQ 5.3 - Как настроить очередь для отклонения дублирующихся сообщений? , это также может помочь.

4 голосов
/ 08 августа 2013

Добавлена ​​поддержка удаления дубликатов сообщений, запакованных в транспорт ActiveMQ. См. Значения конфигурации auditDepth и auditMaximumProducerNumber в Руководстве по конфигурации подключения .

3 голосов
/ 27 августа 2012

Есть способ заставить ActiveMQ фильтровать дубликаты на основе свойства JMS. это включает в себя написание Activemq Plugin . Базовый фильтр брокера, который отправляет повторяющиеся сообщения в очередь deadletter, будет выглядеть так: 

import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.activemq.broker.Broker;
import org.apache.activemq.command.Message;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.broker.ProducerBrokerExchange;

public class DuplicateFilterBroker extends BrokerFilter {
    String messagePropertyName;
    boolean switchValue;

    public DuplicateFilterBroker(Broker next, String messagePropertyName) {
        super(next);
        this.messagePropertyName = messagePropertyName;
    }

    public boolean hasDuplicate(String propertyValue){
        switchValue = propertyValue;
        return switchValue;
    }

    public void send(ProducerBrokerExchange producerExchange, Message msg) throws Exception { 
        ActiveMQMessage amqmsg = (ActiveMQMessage)msg; 
        Object msgObj = msg.getMessage(); 
        if (msgObj instanceof javax.jms.Message) { 
            javax.jms.Message jmsMsg = (javax.jms.Message) msgObj; 
            if (!hasDuplicate(jmsMsg.getStringProperty(messagePropertyName))) {
                super.send(producerExchange, msg);
            }
            else {
               sendToDeadLetterQueue(producerExchange.getConnectionContext(), msg);
            } 
        }
    }  
}
0 голосов
/ 13 декабря 2016

Кажется, способ, предложенный в вопросе, работает и для ActiveMQ (2016/12). См. Руководство activemq-artemis . Это требует, чтобы производитель установил определенное свойство в сообщении. 

Message jmsMessage = session.createMessage();
String myUniqueID = "This is my unique id";   // Could use a UUID for this
message.setStringProperty(HDR_DUPLICATE_DETECTION_ID.toString(), myUniqueID);

Однако класс, содержащий свойство, отличается: org.apache.activemq.artemis.core.message.impl.HDR_DUPLICATE_DETECTION_ID и значение свойства _AMQ_DUPL_ID.

...