Интеграция с Spring: сложность транзакции между двумя активаторами

Question

У меня есть этот вариант использования.

Первая цепочка:

<int:chain input-channel="inserimentoCanaleActivate" output-channel="inserimentoCanalePreRouting">      
    <int:service-activator ref="inserimentoCanaleActivator" method="activate" />                
</int:chain>

Это относительный код:

@Override
@Transactional(propagation = Propagation.REQUIRES_NEW)
public EventMessage<ModificaOperativitaRapporto> activate(EventMessage<InserimentoCanale> eventMessage) {
    ...
    // some Database changes
    dao.save(myObject);
}

Все отлично работает.

Тогда у меня есть другая цепочка:

<int:chain id="onlineCensimentoClienteChain" input-channel="ONLINE_CENSIMENTO_CLIENTE" output-channel="inserimentoCanaleActivate">
    <int:service-activator ref="onlineCensimentoClienteActivator" method="activate" />
    <int:splitter expression="payload.getPayload().getCanali()" />
</int:chain>

И относительный активатор:

@Override
public EventMessage<CensimentoCliente> activate(EventMessage<CensimentoCliente> eventMessage) {
    ...
    // some Database changes
    dao.save(myObject);
}

Полезная нагрузка CensimentoCliente, как описано ниже, имеет List полезной нагрузки первогоцепочка, поэтому с помощью сплиттера я делю список и повторно использую код первой цепочки.

public interface CensimentoCliente extends Serializable {

    Collection<? extends InserimentoCanale> getCanali();

    void setCanali(Collection<? extends InserimentoCanale> canali);
    ...
}

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

Цель состоит в том, чтобы модификации базы данных двух цепочек были частью одной транзакции.

Любая помощь?

С уважением, Массимо

Answer 1

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

public class TransactionalChannel extends AbstractSubscribableChannel {

    private final MessageDispatcher dispatcher = new UnicastingDispatcher();
    private final TransactionTemplate transactionTemplate;

    TransactionalChannel(TransactionTemplate transactionTemplate) {
        this.transactionTemplate = transactionTemplate;
    }

    @Override
    protected boolean doSend(final Message<?> message, long timeout) {
        return transactionTemplate.execute(new TransactionCallback<Boolean>() {
            @Override
            public Boolean doInTransaction(TransactionStatus status) {
                return getDispatcher().dispatch(message);
            }
        });
    }

    @Override
    protected MessageDispatcher getDispatcher() {
        return dispatcher;
    }

}

В своем XML вы можете определить свой канал и шаблон транзакции и ссылаться на свой пользовательский канал так же, как и на любой другой канал:

    <bean id="transactionalChannel" class="com.stackoverflow.TransactionalChannel">
        <constructor-arg>
           <bean class="org.springframework.transaction.support.TransactionTemplate">
              <property name="transactionManager" ref="transactionManager"/>
              <property name="propagationBehavior" value="#{T(org.springframework.transaction.TransactionDefinition).PROPAGATION_REQUIRES_NEW}"/>
          </bean>
       </constructor-arg>
    </bean>

Например, вы можете использовать мост для передачи сообщения по новому каналу:

<int:bridge input-channel="inserimentoCanaleActivate" output-channel="transactionalChannel" /> 
<int:chain input-channel="transactionalChannel" output-channel="inserimentoCanalePreRouting">      
    <int:service-activator ref="inserimentoCanaleActivator" method="activate" />                
</int:chain>

Answer 2

Если у вас есть <service-activator> и @Transactional для метода обслуживания, транзакция будет ограничена только вызовом этого метода. Если вы хотите получить преобразование для всего потока сообщений (или его части), вы должны объявить рекомендацию TX где-то раньше. Если ваши каналы являются прямыми, все вызовы службы будут заключены в одну транзакцию. Самый простой способ выполнить ваши желания, написать простой @Gateway интерфейс с @Transactional и вызывать его с самого начала вашего потока сообщений.

Чтобы уточнить немного относительно транзакций Понимание транзакций в потоках сообщений

Answer 3

Они модифицируют 2 отдельные реляционные базы данных? Если это так, вы смотрите на транзакцию XA. Теперь, если вы выполняете это в не XA-контейнере, таком как tomcat, все это должно быть выполнено в одном потоке, который отслеживается менеджером транзакций - (вам придется добавить диспетчер транзакций, который фактически вызывает эти события). Менеджер транзакций может быть сообщением JMS или средством опроса какого-либо источника данных. Также эта обработка должна выполняться в одном потоке, чтобы Spring мог помочь вам запустить весь процесс в одной транзакции.

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