Блог

Что было бы эквивалентно следующей конфигурации адаптера исходящего канала в spring boot? Предполагая messageChannel , taskExecutor , и synchronizationFactory определены.

     <int:outbound-channel-adapter id="outboundChannelAdapter" channel="messageChannel" ref="handler" method="handle">
        <int:poller task-executor="taskExecutor" fixed-delay="500" receive-timeout="500" max-messages-per-poll="10">
            <int:transactional synchronization-factory="synchronizationFactory" isolation="READ_COMMITTED"/>
        </int:poller>
    </int:outbound-channel-adapter>
 

@ServiceActivator @Poller Похоже, что у с аннотацией нет опции для фабрики синхронизации транзакций.
Для PollerMetadata этого есть опция, но я не уверен, как подключить этот экземпляр к @ServiceActivator .

В этом случае необходима фабрика синхронизации, поскольку это канал на основе базы данных, из которого считывается несколько потоков.

Что-то вроде этого:

 @ServiceActivator(inputChannel = "messageChannel", poller = @Poller("myPollerMetadata"))
public void handle(Message<?> message) { // Or what is your service method signature
    ...
}

@Bean
PollerMetadata myPollerMetadata(Executor taskExecutor, TransactionSynchronizationFactory synchronizationFactory) {

    PollerMetadata poller = new PollerMetadata();
    poller.setTransactionSynchronizationFactory(synchronizationFactory);
    poller.setMaxMessagesPerPoll(10);
    poller.setReceiveTimeout(500);
    poller.setTaskExecutor(taskExecutor);
    poller.setTrigger(new PeriodicTrigger(500));
    return poller;
}
 

Вы также можете подумать о том, чтобы начать изучать Spring Integration Java DSL: https://docs.spring.io/spring-integration/reference/html/dsl.html#java-dsl . Та же конфигурация будет выглядеть следующим образом:

 @Bean
IntegrationFlow myHandlerFlow(Executor taskExecutor, TransactionSynchronizationFactory synchronizationFactory) {
    return IntegrationFlows.from("messageChannel")
              .handle(handler, "handle", 
                            c -> c.poller(p -> p
                                     .fixedDelay(500)
                                     .transactionSynchronizationFactory(synchronizationFactory)
                                     .taskExecutor(taskExecutor)
                                     .receiveTimeout(500)
                                     .maxMessagesPerPoll(10)))
              .get();
}