如何在Webflux应用中使Spring云流消费者？

Question

我有一个基于Webflux的微服务，它有一个简单的反应存储库：

    public interface NotificationRepository extends ReactiveMongoRepository<Notification, ObjectId> {
    }

    @Bean
    public Consumer<KStream<String, Event>> documents(NotificationRepository repository) {
        return input ->
                input.foreach((key, value) -> {
                    LOG.info("Received event, Key: {}, value: {}", key, value);
                    repository.save(initNotification(value)).subscribe();
                });
    }

Native support for reactive programming - since v3.0.0 we no longer distribute spring-cloud-stream-reactive modules and instead relying on native reactive support provided by spring cloud function. For backward compatibility you can still bring spring-cloud-stream-reactive from previous versions.

而且在这个演示视频中，他们提到他们使用project Reactor有反应性编程支持。所以我想有一种方法我只是不知道。你能教我怎么做对吗？

如果这一切听起来太愚蠢，我很抱歉，但我对Spring、Cloud、Stream和reactive编程非常陌生，还没有找到很多描述这方面的文章。

Answer 1

只需将Flux用作已消耗类型，如下所示：

@Bean
public Consumer<Flux<Message<Event>>> documents(NotificationRepository repository) {
    return input ->
            input
             .map(message-> /*map the necessary value like:*/ message.getPayload().getEventValue())
             .concatMap((value) -> repository.save(initNotification(value)))
             .subscribe();
}

如果使用返回类型为空的函数(函数 >、mono > )而不是使用者，那么框架可以自动订阅。对于consumer，您必须手动订阅，因为框架没有对流的引用。但是在consumer情况下，您订阅的不是存储库，而是整个流，这是可以的。