Spring Cloud Stream消费者交换连接重试

Question

我有带有反应式 Spring Cloud Stream 的rabbitmq消费者

@Bean
public <T> Consumer<Flux<Message<Subscription>>> exportConsumer() {
    return new ExportSubscriptionConsumer<>();
}

配置

spring:
cloud:
    function:
        definition: exportConsumer
    stream:
        rabbit:
            bindings:
                exportConsumer-in-0:
                    consumer:
                        bindQueue: true
                        queueNameGroupOnly: true
                        declareExchange: false
                        bindingRoutingKey: smfexportconsumer
        function:
            bindings:
                exportConsumer-in-0:
                    destination: jms.durable.queues
        bindings:
            exportConsumer-in-0:
                group: smfexportconsumer

，下面是我遇到的

Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'exportConsumer-in-0' in vhost '9509bf37-8afc-486e-ac98-61034066671d', class-id=50, method-id=20)

问题，在应用程序消费了几条消息后突然抛出了一个错误。如果我重新启动应用程序，它会工作一段时间，但又出现同样的问题！ 我是否可以选择在此处配置重试，找不到任何此类配置此处 或者 我可以选择配置我的自定义重试逻辑吗？ 请帮忙。

[编辑]

messageFlux
            .doOnNext(s -> logger.info("Got message: {}", s))
            .map(Message::getPayload)
            .flatMap(this::converToCharacters)
            .map(User::getId)
            .flatMap(this::getUser)
           .onErrorContinue((throwable, o) -> logger.error(throwable.getMessage()))

在转换方法内，我有一个 API 调用。

return webClient.get()
            .uri(uriBuilderFactory.builder().path("/authorize").build())
            .retrieve().bodyToMono(String.class)
            .publishOn(Schedulers.immediate())
            .timeout(Duration.ofSeconds(10))
            .map(s1 -> new User(s, s1));

现在我的问题是，如果我在上述端点上超时，我该如何处理？它使用

onErrorContinue

还是我可以选择全局错误处理？

[编辑2] 示例代码此处