性能优化：RxJava异步响应式编程提升响应速度

本文通过引入RxJava，通过异步的方式，以短短30行的代码，解决了一个常见的性能问题，提升界面的响应速度。

作为例子，请看下图：

单线程的问题

上图中，ID，Status，Runtime数据的获取都需要时间。如果用单线程去获取，然后再显示，那么程序的响应时间就是所有时间的总和，用户需要等待很久才能看到界面的更新。下面演示了单线程的问题：

性能优化：RxJava异步响应式编程

我们可以采用多线程并发执行的方法去解决问题。期望的效果如下图：

首先，快速读取部分数据（比如只读ID列到Data对象中），然后加载到界面中，这样界面就能快速响应了，虽然此时表格只有ID列有值。

List<Data> datasOnlyWithId = service.fastLoad(); // 快速读取，由于只读ID，因此较快完成，所以直接在UI线程运行。
tableViewer.setInput(datasOnlyWithId);    // 显示到表格中

然后，通过异步任务并发地加载其他数据，在用户接受的时间内（比如500ms），每加载到一个或多个数据，更新到表格中显示。

为了能尽快完成所有任务，为每一个ID对应的Status的获取过程创建一个并发任务（就有n个任务，n为ID的个数，也就是表格总的行数）：

    Observable<Data> obsvData = Observable.from(datasOnlyWithId);

    Observable<Data> obsvLoadStatus = obsvData.flatMap(data -> 
Observable.<Data>create(subscriber -> {
data.setStatus(dao.loadStatus(data.getId()));  // DAO加载ID对应的Status，需要一些时间
        subscriber.onNext(data);
        subscriber.onCompleted();
      })
.subscribeOn(Schedulers.io())
    );

上面代码中，用from方法从列表datasOnlyWithId创建出Observable对象，它是响应式处理过程的起始点，输入源，事件源。每个事件的数据就是一个只包含ID的Data数据项。

对于每个Data（即表格每行的数据），分配异步任务运行DAO的loadStatus这个耗时的操作。这里要注意到地方是Data这个类的对象会在不同线程中被修改（具体就是其Status域和Runtime域），需要确保其线程安全。

类似的，为每一个ID对应的Run Time的获取过程创建一个并发任务（就有n个任务，n为ID的个数，也就是表格总的行数）：

    Observable<Data> obsvLoadRunTime = obsvData.flatMap(data -> 
Observable.<Data>create(subscriber -> {
data.setRunTime(dao.loadRunTime(data.getId())); // DAO加载ID对应的RunTime，需要一些时间
        subscriber.onNext(data);
        subscriber.onCompleted();
      })
.subscribeOn(Schedulers.io())
    );

以上两类任务合并到一起构成新的可观察对象，以便订阅者统一处理：

Observable.<Data>merge(obsvLoadStatus, obsvLoadRunTime)

每个任务完成的时间都不一样，有的快，有的慢，因此整个界面完全加载的时间就是那个最慢的任务的时间，而不是所有时间的总和，因此就快了 2N 倍。

如果每次只要有数据（比如某个ID对应的Status或者RunTime）获取到都去更新UI，那么UI更新也太频繁（同一行数据可能会更新多次），因为只要不影响用户体验就行。比如每500ms检测一下有哪些数据加载完成的，有的话就批量更新。实现这个要求用RxJava很简单，用buffer方法就可以。

// （接上面的代码，下面也是同样的规则）
.buffer(500, TimeUnit.MILLISECONDS)

在批量更新之前，把重复的数据过滤掉，防止重复更新。用distinct方法即可：

.flatMap(dataList -> Observable.from(dataList).distinct())

以上操作都是在IO线程池中操作。但是GUI框架要求界面的操作只能由UI线程来执行，比如要在表格更新数据显示。GUI作为对以上操作的响应，我们可以用observeOn方法来指定有哪个线程池（线程）来响应。这里感谢zakgof为SWT的Rx绑定提供的开源的rxswt。

.observeOn(SwtScheduler.getInstance())

UI响应的方式是，每当有数据需要更新，就只更新数据对应的那一行；有错误就显示错误信息；全部完成也弹出一个对话框显示。

.subscribe(data -> { // 响应每个事件
LOGGER.debug(data.toString());
tableViewer.refresh(data);
}, ex -> { // 响应处理未捕获的异常
LOGGER.error(ex.getMessage(), ex);
MessageDialog.openError(getShell(), "Error", ex.getMessage());
}, () -> { // 当全部处理完时的响应
showDoneMessage(startTime);
});

运行程序，表格立刻显示出来，一开始只是ID列有数据，但其他数据也在不断的显示出来，而且加载过程界面也不卡。全部数据都显示出来的时间也大大地加快（只取决于最慢那个任务的执行时间，而不是所有任务执行时间的总和），见下面的Demo：

查看打出的log，可见RxJava创建了 14 个并发任务 （数据个数 n = 7, 每个数据有Status和RunTime两个值需要耗时获取，因此需要的并发任务数是 2 n = 14）。

09:38:45.465 [main] DEBUG rx.swt.test.dao.DataDao - loadIdList
09:38:45.846 [RxIoScheduler-4] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 3
09:38:45.846 [RxIoScheduler-6] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 5
09:38:45.846 [RxIoScheduler-2] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 1
09:38:45.846 [RxIoScheduler-5] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 4
09:38:45.846 [RxIoScheduler-9] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 1
09:38:45.846 [RxIoScheduler-3] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 2
09:38:45.846 [RxIoScheduler-7] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 6
09:38:45.846 [RxIoScheduler-8] DEBUG rx.swt.test.dao.DataDao - loadStatus for id 7
09:38:45.846 [RxIoScheduler-10] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 2
09:38:45.846 [RxIoScheduler-11] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 3
09:38:45.846 [RxIoScheduler-12] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 4
09:38:45.846 [RxIoScheduler-13] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 5
09:38:45.846 [RxIoScheduler-14] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 6
09:38:45.846 [RxIoScheduler-15] DEBUG rx.swt.test.dao.DataDao - loadRunTime for id 7
09:38:46.333 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=1, status=Abort, runTime=0]
09:38:46.837 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=2, status=, runTime=864]
09:38:46.837 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=5, status=Running, runTime=0]
09:38:47.837 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=7, status=, runTime=1987]
09:38:48.337 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=3, status=Abort, runTime=2352]
09:38:48.837 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=5, status=Running, runTime=2641]
09:38:48.838 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=7, status=Running, runTime=1987]
09:38:49.337 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=4, status=, runTime=3456]
09:38:49.837 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=4, status=New, runTime=3456]
09:38:49.838 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=6, status=Abort, runTime=0]
09:38:50.337 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=1, status=Abort, runTime=4213]
09:38:50.337 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=2, status=New, runTime=864]
09:38:50.861 [main] DEBUG rx.swt.test.ui.MainComposite - Data [id=6, status=Abort, runTime=5023]
09:38:50.871 [main] DEBUG rx.swt.test.ui.MainComposite - Success to load all the data, total time is 5.406 s

使用RxJava很方便，极大降低编写多线程并发程序的难度，而且代码量还可以这么少 —- 只有区区30行代码！

—— 本博客所有内容均为原创，转载请注明作者和出处 ——–

作者：刘文哲

联系方式：[e-mail]

博客：http://blog.csdn.net/liuwenzhe2008

源码： https://github.com/WenzheLiu/rx.swt.sample