Spring Integration 的 Java DSL 和 H2 适配器

Question

我正在尝试创建与以下配置等效的配置：

<int:channel id="fromdb"/>
<int:service-activator input-channel="fromdb" ref="jdbcMessageHandler" />
<int-jdbc:inbound-channel-adapter
    channel="fromdb" 
    data-source="dataSource"
    query="SELECT * FROM Items WHERE INVENTORY_STATUS = 0"
    update="UPDATE Items SET INVENTORY_STATUS = 1">
    <int:poller fixed-delay="4000" />

在 Spring 集成 Java DSL 中：

@SpringBootApplication
public class DbPollerSimpleDSLApplication {

    @Bean
    public MessageSource<?> jdbcAdapter(DataSource dataSource) {
        JdbcPollingChannelAdapter adapter =
                new JdbcPollingChannelAdapter(dataSource, "SELECT * FROM Items WHERE INVENTORY_STATUS = 0");
        adapter.setUpdateSql("UPDATE Items SET INVENTORY_STATUS = 1");
        return adapter;
    }

    @Bean
    public IntegrationFlow jdbcFlow(MessageSource<?> jdbcAdapter) {
        return IntegrationFlows
                .from(jdbcAdapter, e ->
                        e.poller(p -> p.fixedRate(4000)/*.transactional(transactionManager())*/))
                .channel(c -> c.direct("fromdb"))
                .get();
    }

    /*@Bean
    public MessageChannel fromdb() {
        return new DirectChannel();
    }*/

    public static void main(String[] args) {
        SpringApplication.run(DbPollerSimpleDSLApplication.class, args);
    }
}

导致以下错误：

Error starting ApplicationContext. To display the auto-configuration report re-run your application with 'debug' enabled.
2018-03-09 20:00:06.781 ERROR 23804 --- [           main] o.s.boot.SpringApplication               : Application startup failed

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'jdbcFlow' defined in com.example.DbPollerSimpleDSLApplication: Initialization of bean failed; nested exception is java.lang.IllegalStateException: Could not register object [fromdb] under bean name 'fromdb': there is already object [fromdb] bound
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:564) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:483) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:306) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:230) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:302) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:197) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:761) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:867) ~[spring-context-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:543) ~[spring-context-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:122) ~[spring-boot-1.5.10.RELEASE.jar:1.5.10.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693) [spring-boot-1.5.10.RELEASE.jar:1.5.10.RELEASE]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360) [spring-boot-1.5.10.RELEASE.jar:1.5.10.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:303) [spring-boot-1.5.10.RELEASE.jar:1.5.10.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1118) [spring-boot-1.5.10.RELEASE.jar:1.5.10.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1107) [spring-boot-1.5.10.RELEASE.jar:1.5.10.RELEASE]
    at com.example.DbPollerSimpleDSLApplication.main(DbPollerSimpleDSLApplication.java:43) [classes/:na]
Caused by: java.lang.IllegalStateException: Could not register object [fromdb] under bean name 'fromdb': there is already object [fromdb] bound
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.registerSingleton(DefaultSingletonBeanRegistry.java:130) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.registerSingleton(DefaultListableBeanFactory.java:936) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.integration.dsl.config.IntegrationFlowBeanPostProcessor.registerComponent(IntegrationFlowBeanPostProcessor.java:291) ~[spring-integration-java-dsl-1.2.3.RELEASE.jar:na]
    at org.springframework.integration.dsl.config.IntegrationFlowBeanPostProcessor.processStandardIntegrationFlow(IntegrationFlowBeanPostProcessor.java:184) ~[spring-integration-java-dsl-1.2.3.RELEASE.jar:na]
    at org.springframework.integration.dsl.config.IntegrationFlowBeanPostProcessor.postProcessBeforeInitialization(IntegrationFlowBeanPostProcessor.java:100) ~[spring-integration-java-dsl-1.2.3.RELEASE.jar:na]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyBeanPostProcessorsBeforeInitialization(AbstractAutowireCapableBeanFactory.java:409) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1620) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:555) ~[spring-beans-4.3.14.RELEASE.jar:4.3.14.RELEASE]
    ... 15 common frames omitted

以下替代配置可以正常工作：

@SpringBootApplication
public class DbPollerSimpleApplication {

    @Bean
    public MessageChannel fromdb() {
        return new DirectChannel();
    }

    @Bean
    public PollerMetadata poller(PlatformTransactionManager transactionManager) {
        PeriodicTrigger trigger = new PeriodicTrigger(4000);
        trigger.setFixedRate(true);

        /*MatchAlwaysTransactionAttributeSource attributeSource = new MatchAlwaysTransactionAttributeSource();
        attributeSource.setTransactionAttribute(new DefaultTransactionAttribute());
        TransactionInterceptor interceptor = new TransactionInterceptor(transactionManager, attributeSource);*/

        PollerMetadata poller = new PollerMetadata();
        poller.setTrigger(trigger);
        //poller.setAdviceChain(Collections.singletonList(interceptor));
        return poller;
    }

    @Bean
    @InboundChannelAdapter(value = "fromdb", poller = @Poller("poller"))
    public MessageSource<?> counterMessageSource(DataSource dataSource) {
        JdbcPollingChannelAdapter adapter =
                new JdbcPollingChannelAdapter(dataSource, "SELECT * FROM Items WHERE INVENTORY_STATUS = 0");
        adapter.setUpdateSql("UPDATE Items SET INVENTORY_STATUS = 1");
        return adapter;
    }


    public static void main(String[] args) {
        SpringApplication.run(DbPollerSimpleApplication.class, args);
    }
}

我在这里错过了什么？

更新：

@SpringBootApplication
public class DbPollerSimpleDSLApplication {

    @Bean
    public MessageSource<?> jdbcAdapter(DataSource dataSource) {
        JdbcPollingChannelAdapter adapter =
                new JdbcPollingChannelAdapter(dataSource, "SELECT * FROM Items WHERE INVENTORY_STATUS = 0");
        adapter.setUpdateSql("UPDATE Items SET INVENTORY_STATUS = 1");
        return adapter;
    }

    @Bean
    public IntegrationFlow jdbcFlow(MessageSource<?> jdbcAdapter) {
        return IntegrationFlows
                .from(jdbcAdapter, e -> e.poller(p -> p.fixedRate(4000)/*.transactional(transactionManager())*/))
                .channel("fromdb")
                .get();
    }

    /*@Bean
    public MessageChannel fromdb() {
        return new DirectChannel();
    }*/

    public static void main(String[] args) {
        SpringApplication.run(DbPollerSimpleDSLApplication.class, args);
    }
}

Answer 1

您似乎遇到了此处描述的问题：https://docs.spring.io/spring-integration/docs/5.0.3.RELEASE/reference/html/java-dsl.html#java-dsl-channels

查看重要段落

注意通过来自不同IntegrationFlow s 的MessageChannels 工厂使用相同的内联通道定义。即使 DSL 解析器将不存在的对象注册为 bean，它也无法从不同的 IntegrationFlow 容器中确定相同的对象 (MessageChannel)。这是错误的：

所以，如果你有：

@Bean
public MessageChannel fromdb() {

您应该在IntegrationFlow 中只使用它的 bean 名称，因此：

.channel("fromdb")

这与您提到的 @InboundChannelAdapter(value = "fromdb" 完全相同。