Spring Cloud Stream Responsive-如果是反应流管道，如何进行错误处理？

Question

如何对反应流管道进行错误处理。喜欢

• 应用程序错误处理（例如：errorChannel）
• 系统错误处理（使用DLQ、再处理等）

当前文档仅描述了非反应性管道的错误处理。https://docs.spring.io/spring-cloud-stream/docs/Fishtown.BUILD-SNAPSHOT/reference/htmlsingle/#_application_error_handling

通过为错误处理场景提供简单的配置，Spring cloud stream为用户提供了非常简单的操作。如果相同的错误处理用例（使用相同的配置）也适用于反应流管道，那就太好了。用例和各自的配置详细信息如下：

• @StreamListner（"errorChannel"）全局错误处理的注释
• @KafkaListener（id="bar"，主题="reactive-stream-error-subject"）
• 为DLQ配置并生成错误主题的失败消息spring.cloud.stream.kafka.bindings.input.consumer.enableDlq=truespring.cloud.stream.kafka.bindings.input.consumer.dlqName=reactive-stream-error-subject

文档中的示例可以很好地处理spring cloud stream，但同样的情况也会给反应式管道带来错误。这方面的任何指导方针都将对社区大有帮助。提前感谢！

@SpringBootApplication
@EnableBinding(Sink.class)
public class ReactiveStreamSinkApplication {

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

@StreamListener
public void receive(@Input(Sink.INPUT) Flux<String> inputFlux) {
    inputFlux.subscribe(System.out::println);
    throw new RuntimeException("BOOM!");
}

@StreamListener("errorChannel")
public void error(Message<?> message) {
    // log the error msg
    System.out.println("Handling ERROR: " + message);
}

@KafkaListener(id="bar", topics = "reactive-stream-error-topic")
public void error(String in) {
    System.out.println(in + " from DLQ");
}
}

Answer 1

很抱歉，回复太晚了。

首先，您的代码中有一个问题，您正在处理如何从反应流中抛出异常。基本上，您处理的是一个声明性处理程序，它的处理方式非常不同。在代码中，在启动和初始化期间，receive方法只会被调用一次。因此，从中引发异常与流处理过程中引发的异常完全不同，您所查询的错误处理机制就是为这个机制而设计的。但是

除此之外。

随着Spring Cloud Function编程模型的引入，我们正在考虑将注意力从反应式模块转移到一起，因为Spring Cloud Function已经提供了对反应式编程模型的支持。所以考虑以下几点：

@SpringBootApplication
@EnableBinding(Sink.class)
public class ReactiveStreamSinkApplication {

    public static void main(String[] args) {
        SpringApplication.run(ReactiveStreamSinkApplication.class,
        "--spring.cloud.stream.function.definition=myconsumer");
    }

    @Bean
    public Consumer<Flux<String>> myconsumer() {
        return stream -> stream.subscribe(value -> {
            if ("foo".equals(value)) {
                throw new RuntimeException("BOOM!");
            }
            System.out.println("Received value: " + value);
        });
    }

    @StreamListener("errorChannel")
    public void error(Message<?> message) {
        // log the error msg
        System.out.println("Handling ERROR: " + message);
    }
}

试试看，让我们知道。