Reactive Streams：Spring WebFlux - 订阅现有的发布者答案

【问题标题】：Reactive Streams: Spring WebFlux - subscribe to existing publisherReactive Streams：Spring WebFlux - 订阅现有的发布者
【发布时间】：2018-09-21 03:44:30
【问题描述】：

我目前正在将我们现有的 Spring 异步 REST 架构迁移到 Spring 的新 WebFlux 库，并且有一个关于加入多个请求以便它们可以监听相同的已发布响应的问题。

用例如下：

客户端 A 连接到我们的网络服务器并请求数据
我们访问缓存以检查是否有数据
我们没有，所以我们去检索这些数据（客户端 A 已订阅并等待响应）
客户端 B 连接到我们的 Web 服务器并请求相同的数据（访问相同的端点）
我们检查缓存，数据仍然不存在
由于我们已经在为客户端 A 获取此数据，因此我们不想发出另一个请求，但是我们也不想拒绝客户端 B。客户端 B 应该能够监听相同的信息

客户端 B 如何订阅客户端 A 正在等待的相同响应流？

【问题讨论】：

你能提供代码 sn-ps 来展示缓存库 API 的样子吗？另外，这里的预期行为是什么？我假设缓存可能不包含要发送给客户端的原始响应，所以它是关于在服务层共享缓存数据，而不一定是在 Web 层？
我实际上只是想知道反应模式是否可能。示例中使用缓存来说明为什么请求可能不会立即得到响应。我想知道两个请求是否到达同一个端点，WebFlux 是否可以加入它们并允许它们都等待相同的响应？显然，经典的 REST 和 Spring 无法做到这一点，但我希望反应式流可以实现这种行为。

标签： java spring spring-webflux reactive reactive-streams

【解决方案1】：

“客户端 A 已订阅并等待响应” 我想请求被编码为Mono 并且客户端 A sibscribes 字面意思：

Subscriber<Response> clientA = ... Mono<Response> request = makeRequest(...); request.subscribe(clientA);

那么clientB应该以同样的方式订阅：

Subscriber<Response> clientB = ... request.subscribe(clientB);

此外，缓存不应该包含以前保存的响应数据，而是请求本身，类型为Mono<Response>。然后，如果在缓存中找到这样的请求，新客户端只需订阅它，而不管该请求是否已经完成。

【讨论】：

本例中的“缓存”只是一个非常轻量级的 Hazelcast IMap 数据结构。当且仅当所述数据不存在时，它才会填充来自外部源的数据。我想要做的是强制外部数据调用（和缓存填充）不会以最优雅的方式发生两次。已经有多种方法可以解决这个问题，但我想知道是否可以使用 WebFlux 以简单的方式解决它——我的想法是两个请求监听相同的事件序列。我无法删除现有的 IMap，但您认为我应该添加一个包含 Mono 的新 IMap？
就像处理器有多个级别的缓存一样，您可以为当前请求添加另一个级别的缓存，包含 Mono，而无需触及 IMap。这可以是简单的 HashMap。
有趣的想法。谢谢。
您可以简单地将接收到的响应从 Mono 缓存移动到主缓存：request.subscribe(response)->{ imapCache.put(response); responseCache.remove(request); }