Блог

Я использую Camel для загрузки сообщений из очереди утешения и с трудом понимаю управление транзакциями.

Поток выглядит следующим образом: сообщение загружается из очереди утешения и затем помещается в одну из многих очередей утешения (называемых промежуточными) в зависимости от некоторой логики. Из этих промежуточных очередей сообщение направляется на процессор (bean) с использованием компонента SEDA и, наконец, отправляется в другую очередь утешения.

Я настроил транзакцию как

 <bean id="propagationReqd" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="jmsTransactionManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
</bean>
  

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

Не удалось создать транзакцию JMS; вложенным исключением является исключение com.solacesystems.jms.ConfigurationException: транзакционные сеансы или сеансы XA не поддерживаются прямым транспортом

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

Конфигурация маршрутизации:

 <camel:route id="msg.router">
    <camel:from uri="{{in.solace.queue}}" />
    <camel:transacted ref="propagationReqd" />
    <camel:to uri="direct:msgSequencer" />
</camel:route>

<camel:route id="msg.processor">
    <camel:from uri="direct:msgSequencer" />
    <camel:transacted ref="propagationReqd" />
    <camel:process ref="sequencer" />
    <camel:choice>
        <camel:when>
            <camel:simple>${headers.MsgId} == '0'</camel:simple>
            <camel:to uri="{{stage.solace.queue.0}}" />
        </camel:when>
        <camel:when>
            <camel:simple>${headers.MsgId} == '1'</camel:simple>
            <camel:to uri="{{stage.solace.queue.1}}" />
        </camel:when>
        ...
        ...
        ...
    </camel:choice>
</camel:route>

<camel:route id="msg.seda.0">
    <camel:from uri="{{stage.solace.queue.0}}" />
    <camel:transacted ref="propagationReqd" />
    <camel:to uri="seda:processor.0" />
</camel:route>

<camel:route id="msg.seda.1">
    <camel:from uri="{{stage.solace.queue.1}}" />
    <camel:transacted ref="propagationReqd" />
    <camel:to uri="seda:processor.1" />
</camel:route> 

<camel:route id="msg.process.0">
    <camel:from uri="seda:processor.0?concurrentConsumers=4amp;amp;waitForTaskToComplete=Neveramp;amp;purgeWhenStopping=true" />
    <camel:transacted ref="propagationReqd" />
    <camel:process ref="msgProcessor" />
    <camel:to uri="{{final.queue}}" />
</camel:route>

<camel:route id="msg.process.1">
    <camel:from uri="seda:processor.1?concurrentConsumers=4amp;amp;waitForTaskToComplete=Neveramp;amp;purgeWhenStopping=true" />
    <camel:transacted ref="propagationReqd" />
    <camel:process ref="msgProcessor" />
    <camel:to uri="{{final.queue}}" />
</camel:route>
  

Признателен, если кто-нибудь может указать, что я делаю не так?

Заранее спасибо.

1. Ошибка звучит как проблема конфигурации с настройкой фабрики соединений Solace.

2. К вашему сведению: конечные точки seda не могут учитываться в транзакции, если они асинхронны.

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

Не удалось создать транзакцию JMS; вложенным исключением является исключение com.solacesystems.jms.ConfigurationException: транзакционные сеансы или сеансы XA не поддерживаются прямым транспортом