定义 Spring 集成中内容扩充器异常时的回退行为

Question

我用弹簧集成定义了一个很好的流程，在名义情况下，它就像我想要的那样工作。

但是，我还没有找到一种方法来定义处理错误的行为(即将输入行标记为失败(

这是我得到的错误：

 [ask-scheduler-2] cMessagingTemplate$TemporaryReplyChannel : Reply message received but the receiving thread has exited due to an exception while sending the request message:ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: error occurred in message handler [outboundGateway]; nested exception is...

这是我的流配置(简化(：

@Configuration
@EnableIntegration
public class IntegrationConfig {

    @Autowired
    private DataSource dataSource;
    @Bean
    public IntegrationFlow mainFlow() {
        //noinspection unchecked
        return IntegrationFlows.from(jdbcMessageSource(),
                c -> c.poller(Pollers.fixedRate(5000)
                        .transactional(transactionManager())))
                .split()
                .enrich(e -> e
                        .requestChannel(subChannel())
                        .requestPayload(Message::getPayload)
                        .propertyExpression("fooId", "payload.id"))
                .handle(barHandler())
                .get();
    }
    @Bean
    public IntegrationFlow enricherFlow() {
        return IntegrationFlows.from(subChannel())
                .handle(outboundGateway())
                .get();
    }
    @Bean
    public MessageChannel subChannel() {
        return new DirectChannel();
    }
    @Bean
    public MessageSource<Object> jdbcMessageSource() {
        return new JdbcPollingChannelAdapter(this.dataSource,
                "select * from FOO");
    }
    @Bean
    public JdbcOutboundGateway outboundGateway() {
        ...
    }
    @Bean
    public JdbcMessageHandler barHandler() {
        return new JdbcMessageHandler(dataSource,
                "INSERT INTO BAR ...");
    }

    @Bean
    public PlatformTransactionManager transactionManager() {
        return new DataSourceTransactionManager(dataSource);
    }
}

我本以为我可以在默认错误频道上出现错误并在那里执行必要的任务，但我还没有找到完成这项工作的方法。

任何想法和帮助都非常感谢！

Answer 1

你知道看起来我们已经错过了在Java DSL中公开EnricherSpec errorChannel - 请随时就此事提出JIRA问题。但无论如何，看起来我们可以访问该属性：

.enrich(e -> {
            ContentEnricher contentEnricher =
                    e.requestChannel(subChannel())
                            .requestPayload(Message::getPayload)
                            .propertyExpression("fooId", "payload.id"))
                            .get().getT2();
            contentEnricher.setErrorChannel(enricherErrorChannel());
        })

现在，您可以将任何.handle()流添加到该enricherErrorChannel()，并在钓鱼时处理ErrorMessage。例如，通常该ErrorMessage包含一个payload作为MessagingException，其中我们有一个failedMessage属性。确切地说，该信息还包含有关您的原始payload和headers的信息：

https://docs.spring.io/spring-integration/docs/5.0.0.RELEASE/reference/html/configuration.html#namespace-errorhandler