Spring cloud stream Rabbitmq绑定器-Spring Cloud函数错误处理

Question

我正在使用具有春季云功能的春季云流兔子活页夹，并定义类似以下的监听器：

public Function<Flux<SomeObject>, Flux<OtherObject>> foo() {
//some code
}

我还将失败的消息重新路由到DLQ。问题是何时发生致命错误，例如org.springframework.messaging.converter.MessageConversionException。它不会像ConditionalRejectingErrorHandler中提到的那样被https://docs.spring.io/spring-amqp/reference/html/#exception-handling处理，并且会永远循环。

是否有办法使ConditionalRejectingErrorHandler起作用？

现在，我通过使用@ServiceActivator(inputChannel = "errorChannel")来解决问题，并亲自处理错误。

依赖关系：

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.4.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
        <dependency>
            <groupId>org.springframework.boot.experimental</groupId>
            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-consul-config</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-hateoas</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-web</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
        </dependency>
</dependencies>

Answer 1

我们长期讨论了用于命令式函数的错误处理和其他功能，以及如何将它们应用于（或什至可以应用于）反应式函数，并尝试了几种不同的方法，但不幸的是，所有这些都归结为阻抗失配。您所描述的方法基于对单个消息的操作。这是命令式消息处理程序（例如Function<String, String>）中的

工作单元