rxjava 循环发送事件

Spring Framework 4.2 GA即将发布，让我们看一下它提供的一些新功能。 引起我注意的一个事件是一个简单的新类SseEmitter ，它是对Spring MVC控制器中容易使用的发送事件的抽象。 SSE是一项技术，可让您在一个HTTP连接内沿一个方向将数据从服务器流式传输到浏览器。 听起来像是websocket可以做什么的子集。 但是，由于它是一个简单得多的协议，因此可以在不需要全双工的情况下使用，例如实时推动股价变化或显示长时间运行的进程。 这将是我们的例子。

假设我们有一个具有以下API的虚拟硬币矿工：

public interface CoinMiner {BigDecimal mine() {//...}
}

每次调用mine()我们都必须等待几秒钟，才能获得大约1个硬币的回报（平均）。 如果要挖掘多个硬币，则必须多次调用此方法：

@RestController
public class MiningController {//...@RequestMapping("/mine/{count}")void mine(@PathVariable int count) {IntStream.range(0, count).forEach(x -> coinMiner.mine());}}

这项工作，我们可以请求/mine/10和mine()方法将执行10次。 到目前为止，一切都很好。 但是挖掘是一项占用大量CPU的任务，将计算分散到多个内核将是有益的。 此外，即使使用并行化，我们的API端点也相当慢，我们必须耐心等待直到所有工作完成而没有任何进度通知。 让我们首先修复并行性–但是，由于并行流无法控制底层线程池，因此我们来使用显式的ExecutorService ：

@Component
class CoinMiner {CompletableFuture<BigDecimal> mineAsync(ExecutorService executorService) {return CompletableFuture.supplyAsync(this::mine, executorService);}//...}

客户端代码必须显式提供ExecutorService （只是设计选择）：

@RequestMapping("/mine/{count}")
void mine(@PathVariable int count) {final List<CompletableFuture<BigDecimal>> futures = IntStream.range(0, count).mapToObj(x -> coinMiner.mineAsync(executorService)).collect(toList());futures.forEach(CompletableFuture::join);
}

首先多次调用mineAsync ，然后（作为第二阶段）等待所有期货完成并join ，这mineAsync重要。 很容易写：

IntStream.range(0, count).mapToObj(x -> coinMiner.mineAsync(executorService)).forEach(CompletableFuture::join);

但是，由于Java 8中流的惰性，该任务将按顺序执行！ 如果您还不习惯流的懒惰，请始终从下至上阅读它们：我们要求join一些将来的内容，以便流上升并只调用一次mineAsync() （惰性！），并将其传递给join() 。 当join()完成时，它再次上升并要求另一个Future 。 通过使用collect()我们强制所有mineAsync()执行，开始所有异步计算。 稍后，我们等待每一个。

介绍

现在该变得更具React性了（我说过了）。 控制器可以返回SseEmitter的实例。 从处理程序方法return后，容器线程将被释放并可以处理更多即将到来的请求。 但是连接没有关闭，客户端一直在等待！ 我们应该做的是保留对SseEmitter实例的引用，并在以后从另一个线程调用其send()和complete方法。 例如，我们可以启动一个长时间运行的进程，并保持send()从任意线程进行进度。 完成该过程后，我们complete() SseEmitter ，最后关闭HTTP连接（至少从逻辑SseEmitter ，请记住Keep-alive ）。 在下面的示例中，我们有一堆CompletableFuture ，当每个CompletableFuture完成时，我们只需将1发送给客户端（ notifyProgress() ）。 当所有期货都完成后，我们完成流（ thenRun(sseEmitter::complete) ），关闭连接：

@RequestMapping("/mine/{count}")
SseEmitter mine(@PathVariable int count) {final SseEmitter sseEmitter = new SseEmitter();final List<CompletableFuture<BigDecimal>> futures = mineAsync(count);futures.forEach(future ->future.thenRun(() -> notifyProgress(sseEmitter)));final CompletableFuture[] futuresArr = futures.toArray(new CompletableFuture[futures.size()]);CompletableFuture.allOf(futuresArr).thenRun(sseEmitter::complete);return sseEmitter;
}private void notifyProgress(SseEmitter sseEmitter) {try {sseEmitter.send(1);} catch (IOException e) {throw new RuntimeException(e);}
}private List<CompletableFuture<BigDecimal>> mineAsync(@PathVariable int count) {return IntStream.range(0, count).mapToObj(x -> coinMiner.mineAsync(executorService)).collect(toList());
}

调用此方法将产生以下响应（注意Content-Type ）：

< HTTP/1.1 200 OK
< Content-Type: text/event-stream;charset=UTF-8
< Transfer-Encoding: chunked
<
data:1data:1data:1data:1* Connection #0 to host localhost left intact

稍后我们将学习如何在客户端解释这种响应。 现在暂时让我们整理一下设计。

与引进RxJava

上面的代码有效，但是看起来很混乱。 我们实际上有一系列事件，每个事件都代表计算的进度。 计算最终完成，因此流也应发出信号结束。 听起来就像是Observable ！ 我们从重构CoinMiner开始，以返回Observable<BigDecimal ：

Observable<BigDecimal> mineMany(int count, ExecutorService executorService) {final ReplaySubject<BigDecimal> subject = ReplaySubject.create();final List<CompletableFuture<BigDecimal>> futures = IntStream.range(0, count).mapToObj(x -> mineAsync(executorService)).collect(toList());futures.forEach(future ->future.thenRun(() -> subject.onNext(BigDecimal.ONE)));final CompletableFuture[] futuresArr = futures.toArray(new CompletableFuture[futures.size()]);CompletableFuture.allOf(futuresArr).thenRun(subject::onCompleted);return subject;
}

每当mineMany()返回的事件出现在Observable ，我们就mineMany()那么多硬币。 当所有期货都完成后，我们也完成了交易。 在实现方面，这看起来还没有改善，但是从控制器的角度来看，它有多干净：

@RequestMapping("/mine/{count}")
SseEmitter mine(@PathVariable int count) {final SseEmitter sseEmitter = new SseEmitter();coinMiner.mineMany(count, executorService).subscribe(value -> notifyProgress(sseEmitter),sseEmitter::completeWithError,sseEmitter::complete);return sseEmitter;
}

调用coinMiner.mineMany()我们只需订阅事件。 事实证明Observable和SseEmitter方法匹配1：1。 这里发生的事情是不言自明的：启动异步计算，每当后台计算发出任何进度信号时，将其转发给客户端。 好的，让我们回到实现上。 由于我们将CompletableFuture和Observable混合使用，因此看起来很混乱。 我已经描述了如何仅使用一个元素将CompletableFuture转换为Observable 。 这是一个概述，包括rx.Single从RxJava 1.0.13开始发现的rx.Single抽象（此处未使用）：

public class Futures {public static <T> Observable<T> toObservable(CompletableFuture<T> future) {return Observable.create(subscriber ->future.whenComplete((result, error) -> {if (error != null) {subscriber.onError(error);} else {subscriber.onNext(result);subscriber.onCompleted();}}));}public static <T> Single<T> toSingle(CompletableFuture<T> future) {return Single.create(subscriber ->future.whenComplete((result, error) -> {if (error != null) {subscriber.onError(error);} else {subscriber.onSuccess(result);}}));}}

将这些实用程序运算符放在某个地方，我们可以改善实现并避免混合使用两个API：

Observable<BigDecimal> mineMany(int count, ExecutorService executorService) {final List<Observable<BigDecimal>> observables = IntStream.range(0, count).mapToObj(x -> mineAsync(executorService)).collect(toList());return Observable.merge(observables);
}Observable<BigDecimal> mineAsync(ExecutorService executorService) {final CompletableFuture<BigDecimal> future = CompletableFuture.supplyAsync(this::mine, executorService);return Futures.toObservable(future);
}

RxJava有一个内置的运算符，用于将多个Observable合并为一个，我们的每个基础Observable发出一个事件。

深入研究RxJava运算符

让我们使用RxJava的功能来稍微改善流式传输。

scan（）

当前，每次我们开采一枚硬币时，我们都会send(1)客户端send(1)事件。 这意味着每个客户都必须跟踪其已经收到的硬币数量，以便计算总的计算数量。 如果服务器总是发送总金额而不是增量，那就太好了。 但是，我们不想更改实现。 事实证明，使用Observable.scan()运算符非常简单：

@RequestMapping("/mine/{count}")
SseEmitter mine(@PathVariable int count) {final SseEmitter sseEmitter = new SseEmitter();coinMiner.mineMany(count, executorService).scan(BigDecimal::add).subscribe(value -> notifyProgress(sseEmitter, value),sseEmitter::completeWithError,sseEmitter::complete);return sseEmitter;
}private void notifyProgress(SseEmitter sseEmitter, BigDecimal value) {try {sseEmitter.send(value);} catch (IOException e) {e.printStackTrace();}
}

scan()运算符接收上一个事件和当前事件，并将它们组合在一起。 通过应用BigDecimal::add我们只需将所有数字相加即可。 例如1、1 + 1，（1 + 1）+ 1，依此类推。 scan()类似于flatMap() ，但保留中间值。

用sample()采样

可能是因为我们的后端服务产生了太多的进度更新，我们无法使用。 我们不想给客户端增加不相关的更新并饱和带宽。 每秒最多发送两次更新听起来很合理。 幸运的是，RxJava也有一个内置的运算符：

Observable<BigDecimal> obs = coinMiner.mineMany(count, executorService);
obs.scan(BigDecimal::add).sample(500, TimeUnit.MILLISECONDS).subscribe(//...);

sample()将定期查看底层流，并仅发出最新的项，并丢弃中间项。 幸运的是，我们使用scan()即时聚合了项目，因此我们不会丢失任何更新。

window() –恒定的发射间隔

不过有一个陷阱。 如果在选定的500毫秒内没有新内容出现， sample()将不会两次发出相同的项目。 很好，但是请记住我们正在通过TCP / IP连接推送这些更新。 最好定期将更新发送给客户端，即使在此期间什么也没发生–只是为了保持连接的正常运行，就像ping 。 可能有多种方法可以实现此要求，例如，涉及timeout()运算符。 我选择使用window()运算符每500毫秒对所有事件进行分组：

Observable<BigDecimal> obs = coinMiner.mineMany(count, executorService);
obs.window(500, TimeUnit.MILLISECONDS).flatMap(window -> window.reduce(BigDecimal.ZERO, BigDecimal::add)).scan(BigDecimal::add).subscribe(//...);

这是一个棘手的问题。 首先，我们将所有进度更新分组在500毫秒的窗口中。 然后，我们使用reduce来计算在此时间段内开采的硬币的总数（类似于scan() ）。 如果在此期间未开采任何硬币，我们只需返回ZERO 。 最后，我们使用scan()汇总每个窗口的小计。 我们不再需要sample()因为window()确保每500毫秒发出一个事件。

客户端

JavaScript中有很多SSE用法示例，因此为您提供一种调用我们的控制器的快速解决方案：

var source = new EventSource("/mine/10");
source.onmessage = function (event) {console.info(event);
};

我相信SseEmitter是Spring MVC的一项重大改进，它将使我们能够编写更健壮和更快的Web应用程序，需要即时的单向更新。

翻译自: https://www.javacodegeeks.com/2015/08/server-sent-events-with-rxjava-and-sseemitter.html

rxjava 循环发送事件