java - 为事件循环创建一个可观察对象

Question

我最近阅读了一些关于 rx-java 的内容。我想知道该框架是否符合线程间通信系统的要求。我正在使用 Java 编写的 REST 服务器。每次 PUT/POST 某些资源时，我都想使用工作线程池进行一些计算。但是，我仍然希望能够监控请求，也许可以打印出一些统计数据。本质上，我想要一个 Observable，这样我就可以使用多个 Observer 以灵活的方式处理请求。

我的问题是，如何创建一个合适的Observable？我见过的大多数指南都涉及对 Observable 的操作，例如映射等。Obervable 主要是从集合或整数范围创建的。在任何情况下，似乎都不可能将新值推送到创建的 Observables。显然，保持这种灵 active 的唯一方法是使用 Observable.create。然而，这似乎是相当低级的。我将不得不为每个新订阅者实现一个队列列表，并对每个订阅者执行 synchronized 推送。这真的有必要吗？或者类似的东西已经在 rx-java 中实现了吗？

Answer 1

您正在寻找的是 Subject .它们充当观察者和可观察对象。例如，ReplaySubject 将向所有订阅者重播发送给它的所有事件。

Subject<String> replaySubject = ReplaySubject.create();
replaySubject.subscribe(s -> System.out.println(s));

// elsewhere...

replaySubject.onNext("First");
replaySubject.onNext("Second");
replaySubject.onComplete();