反应式存储库答案

【问题标题】：Reactive repository反应式存储库
【发布时间】：2018-03-15 09:32:14
【问题描述】：

我尝试创建一些基于响应式堆栈（Reactor + WebFlux）的基本 Spring 5 应用程序。

我的下一个目标是实现能够：

保存图书。
查找所有书籍。

我的存储库需要涵盖以下场景：

场景 A：

没有人订阅 FindAll
有人保存了一本书 (id = 1)
Client1 订阅 FindAll
图书 (id=1) 被推送到 Client1（Client1 保持订阅状态，流未完成！）
有人保存了一本书 (id = 2)
图书 (id=2) 被推送到 Client1（Client1 保持订阅状态，流未完成！）

因此，IMO 这个场景是冷源和热源概念的混合。在任何人订阅之前，我们会收集某人保存在我们存储库中某个缓冲区的数据（比如说普通列表）。对于将订阅 FindAll 的所有订阅者，我们需要推送缓冲列表（在他的订阅之前收集）并且不要完成流以允许推送以后的集合更新。

我能够做到这一点，但我仍然在想有没有更简单的方法可以做到这一点？也许 Reactor 项目中有一个解决方案已经涵盖了这种情况？

我的实现：

public class InMemoryBookRepository {

private final Map<String, Book> bookMap = new ConcurrentHashMap<>();
private final UnicastProcessor<Book> processor = UnicastProcessor.create();
private final FluxSink<Book> fluxSink = processor.sink(FluxSink.OverflowStrategy.LATEST);
private final Flux<Book> hotFlux = processor.publish().autoConnect();

@Override
public void save(Book book) {
    bookMap.put(book.getId(), book);
    fluxSink.next(book);
}

@Override
public Flux<Book> findAll() {
    //without fromIterable I cannot push books that where saved before someone subscribed
    return Flux.fromIterable(bookMap.values())
            .concatWith(hotFlux)
            //Unfortunately this solution produces duplicates so we need to filter them
            .distinct();
}
}

Ofc，我不能只使用冷发布者-因为流将在发布收集的书籍后完成。出于同样的原因，我不能使用 Hot one，因为我会错过在某人订阅之前制作的元素。

旁注：在我的代码中，我的地图没有任何清理机制，因此它会在某些时候产生异常，但这目前并不重要。

【问题讨论】：

只是想知道为什么不能使用像 mongo 这样的反应式数据库？
好问题。但答案很简单——这个存储库是真实系统模拟的一部分。我不想为了模拟目的而将需要单独维护的任何组件添加到我的应用程序中。但是，如果有任何反应式数据库的内存解决方案 - 我一定会试一试。有吗？

标签： java spring project-reactor

【解决方案1】：

如此简单……不知道为什么我错过了这个漂亮的运算符：https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#replay--

所以基本上删除整个List/Map部分代码并使用replay()而不是publish()

简化示例：

UnicastProcessor<String> processor = UnicastProcessor.create();
FluxSink<String> fluxSink = processor.sink(FluxSink.OverflowStrategy.LATEST);
//change 'publish()' to 'replay()'
Flux<String> hotFlux = processor.publish().autoConnect(); 

hotFlux.subscribe(n -> log.info("1st subscriber: {}", n));

fluxSink.next("one");

hotFlux.subscribe(n -> log.info("2nd subscriber: {}", n));

fluxSink.next("two");

输出publish():

1st subscriber: one
1st subscriber: two
2nd subscriber: two

输出replay():

1st subscriber: one
2nd subscriber: one
1st subscriber: two
2nd subscriber: two

【讨论】：