Camel.Два потока обрабатывают одно и то же сообщение одновременно - PullRequest
Новая
       8

Camel.Два потока обрабатывают одно и то же сообщение одновременно

0 голосов
/ 09 октября 2018

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

У меня есть приложение, которое получает и обрабатывает сообщения от ActiveMq, а затем показывает всплывающие окна в веб-браузере.Через некоторое время работы уведомление перестает работать.И я читаю файлы журнала и вижу, что иногда два верблюжьих нити начинают обрабатывать одно и то же сообщение.Почему это происходит?

Должен ли я синхронизировать метод process(final Exchange exchange)?

Фрагмент файлов журнала:

05.10.2018 21:15:52.859 [Camel (app) thread #8 - JmsConsumer[core.consume]] DEBUG InboundMessageProcessor [3.7.13 (default) /]- Process inbound message...

05.10.2018 21:15:52.859 [Camel (app) thread #11 - JmsConsumer[core.consume]] DEBUG InboundMessageProcessor [3.7.13 (default) /]- Process inbound message...

Мой процессор выглядит так: 

public abstract class BaseProcessor implements Processor {

    private static final String LOG4J_VERSION = "version";

    @Override
    public void process(final Exchange exchange) throws MyException {
        try {
            final Message msg = exchange.getIn();

            final String id = msg.getHeader(CustomJMSFields.UNIQUE_ID, String.class);
            if (eventId != null) {
                MDC.put(CustomJMSFields.UNIQUE_ID, id);
            }

            final String name = msg.getHeader(CustomJMSFields.TEL_NAME, String.class);
            if (name != null) {
                MDC.put("name", name );
            }

            final String stationId = msg.getHeader(CustomJMSFields.TEL_STATION_ID, String.class);
            if (stationId != null) {
                MDC.put("stationId", stationId);
            }

            processMessage(msg);
        } finally {
            MDC.remove(CustomJMSFields.UNIQUE_ID);
            MDC.remove("name");
            MDC.remove("stationId");
            MDC.remove(LOG4J_VERSION);
        }
    }

    public abstract void processMessage(Message message) throws MyException;
}

Любая помощь будет по достоинству оценена.

Вот мои настройки JMS: 

    <bean id="jmsConnectionFactoryParent"
      class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="${amq.broker.url}"/>
</bean>

<bean id="jmsConnectionFactory" parent="jmsConnectionFactoryParent"/>

<bean id="simpleMessageConverter"
      class="org.springframework.jms.support.converter.SimpleMessageConverter">
</bean>

<bean id="pooledConnectionFactory"
      class="org.apache.activemq.pool.PooledConnectionFactory" init-method="start" destroy-method="stop">
    <property name="maxConnections" value="20"/>
    <property name="maximumActiveSessionPerConnection" value="25"/>
    <property name="idleTimeout" value="0"/>
    <property name="connectionFactory" ref="jmsConnectionFactory"/>
</bean>

<bean id="jmsConfig"
      class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="connectionFactory" ref="pooledConnectionFactory"/>
    <property name="concurrentConsumers" value="5"/>
    <property name="preserveMessageQos" value="true"/>
    <property name="messageConverter" ref="simpleMessageConverter"/>
</bean>

<bean id="activemq"
      class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>
...