无输出的Spring云流反应侦听器

Question

我使用的是Reactive Spring Cloud Stream，在创建没有输出的StreamListener时遇到了困难。只要没有收到格式错误的消息，下面的代码就可以工作。当收到格式错误的消息时，流量将关闭。

@StreamListener
public void handleMessage(@Input(MessagingConfig.INPUT) Flux<String> payloads) {
    payloads.flatMap(objectToSave -> reactiveMongoTemplate.insert(objectToSave)).subscribe();
}

如果我理解正确的话，最好让框架订阅流量，而不是手动订阅流量。当侦听器有输出时，这不是一个问题，因为我可以简单地返回流量，如下所示：

@StreamListener
@Output(MessagingConfig.OUTPUT)
public Flux<String> handleMessage(@Input(MessagingConfig.INPUT) Flux<String> payloads) {
    return payloads.flatMap(objectToSave -> reactiveMongoTemplate.insert(objectToSave));
}

框架似乎以一种在返回时不会关闭流量的方式处理坏消息。当侦听器没有指定输出时，是否有任何方法可以让框架处理流量？

Answer 1

考虑切换到使用我们最近采用的Spring Cloud Function(SCF)编程模型。基本上，只要您有最新的代码库(2.1.0.RC4是最新的，几天后发布）就可以了。下面是使用SCF编程模型的代码示例：

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

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

    @Bean
    public Consumer<Flux<String>> consume(){
        return payloads -> payloads.flatMap(objectToSave -> reactiveMongoTemplate.insert(objectToSave)).subscribe();
    }
}

您还可以从类路径中删除reactive模块，因为我们也在考虑一起反对它