反应式存储库

Question

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

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

1. 保存书籍。
2. 查找所有图书。

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

方案 A：

1. 没有人订阅FindAll
2. 有人保存了一本书（id=1）
3. Client1订阅FindAll
4. 书籍（id=1）被推送到Client1（Client1保持订阅状态，流未完成！）
5. 有人保存了一本书（id=2）
6. 书籍（id=2）被推送到Client1（Client1保持订阅状态，流未完成！）

所以，在我看来，这个场景是冷源和热源概念的混合。在任何人订阅之前，我们会收集某人保存在我们存储库中某个缓冲区中的数据（比如普通列表）。对于将订阅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，因为我会错过在某人订阅之前生成的元素。

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

Answer 1

这么简单……不知道为什么错过了这个漂亮的接线员：https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#replay--

所以基本上删除整个List/Map部分代码并使用replay（）而不是发布（）

简化示例:

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");

输出与发布（）：

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

使用replay（）输出：

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