目录
- 第一个例子，性能超好？
- 加上业务的模拟，性能超差
- 加上调度器，不起作用？
- RxJava的线程模型
- Schedulers
- 改造，异步执行
- 另一种解决方案
- 总结
- 参考文档
ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软的架构师Erik Meijer领导的团队开发，在2012年11月开源，Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流，Rx库支持.NET、JavaScript和C++，Rx近几年越来越流行了，现在已经支持几乎全部的流行编程语言了，Rx的大部分语言库由ReactiveX这个组织负责维护，比较流行的有RxJava/RxJS/Rx.NET，社区网站是 reactivex.io。
Netflix参考微软的Reactive Extensions创建了Java的实现RxJava，主要是为了简化服务器端的并发。2013年二月份,Ben Christensen 和 Jafar Husain发在Netflix技术博客的一篇文章第一次向世界展示了RxJava。
RxJava也在Android开发中得到广泛的应用。
ReactiveX
An API for asynchronous programming with observable streams.
A combination of the best ideas from the Observer pattern, the Iterator pattern, and functional programming.
虽然RxJava是为异步编程实现的库，但是如果不清楚它的使用，或者错误地使用了它的线程调度，反而不能很好的利用它的异步编程提到系统的处理速度。本文通过实例演示错误的RxJava的使用，解释RxJava的线程调度模型，主要介绍Scheduler、observeOn和subscribeOn的使用。
本文中的例子以并发发送http request请求为基础，通过性能检验RxJava的线程调度。
一、第一个例子，性能超好？
我们首先看第一个例子:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
public static void testRxJavaWithoutBlocking(int count) throws Exception { CountDownLatch finishedLatch = new CountDownLatch(1); long t = System.nanoTime(); Observable.range(0, count).map(i -> { //System.out.println("A:" + Thread.currentThread().getName()); return 200; }).subscribe(statusCode -> { //System.out.println("B:" + Thread.currentThread().getName()); }, error -> { }, () -> { finishedLatch.countDown(); }); finishedLatch.await(); t = (System.nanoTime() - t) / 1000000; //ms System.out.println("RxJavaWithoutBlocking TPS: " + count * 1000 / t); } |
这个例子是一个基本的RxJava的使用，利用Range创建一个Observable, subscriber处理接收的数据。因为整个逻辑没有阻塞，程序运行起来很快，
输出结果为：
RxJavaWithoutBlocking TPS: 7692307 。
二 、加上业务的模拟，性能超差
上面的例子是一个理想化的程序，没雨任何阻塞。我们模拟一下实际的应用，加上业务处理。
业务逻辑是发送一个http的请求，httpserver是一个模拟器，针对每个请求有30毫秒的延迟。subscriber统计请求结果：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
public static void testRxJavaWithBlocking(int count) throws Exception { URL url = new URL("http://127.0.0.1:8999/"); CountDownLatch finishedLatch = new CountDownLatch(1); long t = System.nanoTime(); Observable.range(0, count).map(i -> { try { HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestMethod("GET"); int responseCode = conn.getResponseCode(); BufferedReader in = new BufferedReader(new InputStreamReader(conn.getInputStream())); String inputLine; while ((inputLine = in.readLine()) != null) { //response.append(inputLine); } in.close(); return responseCode; } catch (Exception ex) { return -1; } }).subscribe(statusCode -> { }, error -> { }, () -> { finishedLatch.countDown(); }); finishedLatch.await(); t = (System.nanoTime() - t) / 1000000; //ms System.out.println("RxJavaWithBlocking TPS: " + count * 1000 / t); } |
运行结果如下：
RxJavaWithBlocking TPS: 29。
@＃¥％％……&！
性能怎么突降呢，第一个例子看起来性能超好啊，http server只增加了一个30毫秒的延迟，导致这个方法每秒只能处理29个请求。
如果我们估算一下， 29*30= 870 毫秒，大约1秒，正好和单个线程发送处理所有的请求的TPS差不多。
后面我们也会看到，实际的确是一个线程处理的，你可以在代码中加入
三、加上调度器，不起作用？
如果你对subscribeOn和observeOn方法有些印象的话，可能会尝试使用调度器去解决：
