RxJava中多线程机制

RxJava简介

是一个在Java虚拟机上实现的响应式扩展库:提供了基于observable序列实现的异步调用及基于事件编程。

多线程

RxJava虽然概念比较容易理解，但上手并不容易，而RxJava多线程的机制非常灵活，更需要深入理解。

请参考这篇文章做深入了解，http://gank.io/post/560e15be2dca930e00da1083

实际测试

我准备了如下代码测试RxJava的线程分配机制：

    ...
    public static void main(String[] args) throws Exception {
        Observable.just("a","b","cc")
            .subscribeOn(Schedulers.newThread())
            .observeOn(Schedulers.newThread())
            .map(input -> {
                log("map method in thread: " + Thread.currentThread().getName());
                return input + " == map process";
            })
            .subscribeOn(Schedulers.newThread())
            .observeOn(Schedulers.newThread())
            .map(input -> {
                log("map2 method in thread: " + Thread.currentThread().getName());
                return input + " == map2 process";
            })
            .subscribeOn(Schedulers.newThread())
            .observeOn(Schedulers.newThread())
            .subscribe(input -> {
                log("subscribe method in thread: " + Thread.currentThread().getName());
                System.out.println(input);
            });

        Thread.sleep(2000);
    }
    ...

输出结果：

map method in thread: RxNewThreadScheduler-5 
map method in thread: RxNewThreadScheduler-5
map2 method in thread: RxNewThreadScheduler-3
map method in thread: RxNewThreadScheduler-5
map2 method in thread: RxNewThreadScheduler-3
map2 method in thread: RxNewThreadScheduler-3
subscribe method in thread: RxNewThreadScheduler-1
a == map process == map2 process
subscribe method in thread: RxNewThreadScheduler-1
b == map process == map2 process
subscribe method in thread: RxNewThreadScheduler-1
cc == map process == map2 process

测试结论

• observeOn决定下游代码运行的线程
• observeOn不定义是默认使用当前运行的线程
• subscribeOn决定最原始数据源发射数据代码运行的线程，而接下来的发射数据代码是由observeOn决定的
• RxJava分配线程时是通过调用链自下而上处理，所以最上面的subscribeOn会覆盖下面的定义，也就是说subscribeOn在调用链中最多定义一次就可以，位置任意

具体应用场景

上面说了那么多，好像并没有什么实际的应用价值，最后，不免俗套的举一个Android开发中的应用场景。

需求: 从网络上下载图片，保存到本地并显示到ImageView中

示例代码如下，用到RxAndroid和Retrofit

retrofitService.getImage(url)
    .subscribeOn(Schedulers.io())
    .map(bitmap -> {
        saveTodisk(bitmap);
        return bitmap;
    })
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

访问网络及保存图片都运行在io线程，显示图片运行在主线程，保证界面不卡顿。

所以RxJava让事情变得简单，不管是多线程控制或者业务处理都可以保持在一个调用链上。既保持逻辑的完整又具有灵活扩展的特性，是不可多得的一个利器。