RxJava之多线程、链式调用、Lambda表达式

灰太狼 2023-10-16 22:37 29阅读 0赞

## 参考文章 ##

[http://www.cnblogs.com/halzhang/p/4458095.html][http_www.cnblogs.com_halzhang_p_4458095.html]  
[http://www.tuicool.com/articles/MjyaUfq][http_www.tuicool.com_articles_MjyaUfq]  
[http://www.tuicool.com/articles/VrQRvur][http_www.tuicool.com_articles_VrQRvur]  
[http://blog.csdn.net/lzyzsd/article/details/41833541][http_blog.csdn.net_lzyzsd_article_details_41833541]

#### RxJava的Guihub地址 ####

[https://github.com/ReactiveX/RxJava][https_github.com_ReactiveX_RxJava]

#### 如果对RxJava还不了解，请先看[这里][Link 1] ####

## 多线程处理 ##

### 添加依赖 ###

为了结果更加的直观，添加RxAndroid的依赖
    dependencies {
        compile 'io.reactivex:rxjava:1.0.15'
        compile 'io.reactivex:rxandroid:1.0.1'
    }

### 设置运行线程 ###

通过subscribeOn()来指定Observer的运行线程，通过observeOn()指定Subscriber的运行线程。  
**默认情况下Observer和Subscriber都在当前调用该代码的线程中运行**

subscribeOn//Observable被观察者运行的线程
    observeOn//Subscriber观察者代码运行的线程

Schedulers.io() //运行至子线程（用于 I/O 操作）  
Schedulers.computation() //运行至子线程（计算工作）  
Schedulers.newThread() //运行至子线程（为任务创建的新线程）  
AndroidSchedulers.mainThread() //运行至主线程 需添加compile ‘io.reactivex:rxandroid:1.0.1’

代码

observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                Log.i(TAG, "create 现在的线程:" + Thread.currentThread().getName());
                subscriber.onNext("hello world!");
                subscriber.onCompleted();
            }
        });
        observable = observable.map(new Func1<String, String>() {
            @Override
            public String call(String s) {
                Log.i(TAG, "map 现在的线程:" + Thread.currentThread().getName());
                return s + "这是尾巴~";
            }
        });
        observable = observable.subscribeOn(Schedulers.io()); //设置Subscriber(观察者)运行在子线程
        observable = observable.observeOn(AndroidSchedulers.mainThread()); //设置Observable(被观察者)运行在主线程
    
        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted 现在的线程:" + Thread.currentThread().getName());
            }
    
            @Override
            public void onError(Throwable e) {
    
            }
    
            @Override
            public void onNext(String s) {
                Log.i(TAG, "onNext 现在的线程:" + Thread.currentThread().getName());
                textView.setText(s);
            }
        };
        observable.subscribe(subscriber);

#### 输出结果 ####

create 现在的线程:RxCachedThreadScheduler-1
    map 现在的线程:RxCachedThreadScheduler-1
    onNext 现在的线程:main
    onCompleted 现在的线程:main

并且Activity的textview显示为”hello world!这是尾巴~”;

#### 终止调用 ####

当调用Observable.subscribe()，会返回一个Subscription对象。这个对象代表了被观察者和订阅者之间的联系。  
可以使用unsubscribe来终止调用

#### 代码 ####

修改以上代码

Subscription subscription = observable.subscribe(subscriber);
        subscription.unsubscribe();
        Log.i(TAG, "被终止:" + subscription.isUnsubscribed());

#### 输出结果 ####

被终止:true

### 链式调用 ###

上面的代码为了方便理解没有使用链式调用，实际上，RxJava完全可以使用链式调用。

#### 代码 ####

以上面的代码为例，使用链式调用后

Subscription subscription = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                Log.i(TAG, "create 现在的线程:" + Thread.currentThread().getName());
                subscriber.onNext("hello world!");
                subscriber.onCompleted();
            }
        }).map(new Func1<String, String>() {
            @Override
            public String call(String s) {
                Log.i(TAG, "map 现在的线程:" + Thread.currentThread().getName());
                return s + "这是尾巴~";
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted 现在的线程:" + Thread.currentThread().getName());
            }
    
            @Override
            public void onError(Throwable e) {
    
            }
    
            @Override
            public void onNext(String s) {
                Log.i(TAG, "onNext 现在的线程:" + Thread.currentThread().getName());
                textView.setText(s);
            }
        });

#### 输出结果 ####

和上面的相同

### 更精简版 ###

如果只是输入数据(使用just)，并且不关心Subscriber的OnError和onCompleted，可以更加精简

Subscription subscription = Observable.just("hello world!").map(new Func1<String, String>() {
            @Override
            public String call(String s) {
                Log.i(TAG, "map 现在的线程:" + Thread.currentThread().getName());
                return s + "这是尾巴~";
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                Log.i(TAG, "onNext 现在的线程:" + Thread.currentThread().getName());
                textView.setText(s);
            }
        });

### 使用lambda ###

Java8中可以使用lambda表达式，通过[retrolambda][]这个gradle插件，在Android中可以支持Lambda。

具体Android Studio上retrolambda的配置可参照[android上的JAVA8：使用retrolambda][android_JAVA8_retrolambda]

#### 代码 ####

Observable.just("hello world!").map(s -> {
                Log.i(TAG, "map 现在的线程:" + Thread.currentThread().getName());
                return s + "这是尾巴~";
    }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(s -> {
                Log.i(TAG, "onNext 现在的线程:" + Thread.currentThread().getName());
                textView.setText(s);
    });

[相关代码在这里][Link 2]

[http_www.cnblogs.com_halzhang_p_4458095.html]: http://www.cnblogs.com/halzhang/p/4458095.html
[http_www.tuicool.com_articles_MjyaUfq]: http://www.tuicool.com/articles/MjyaUfq
[http_www.tuicool.com_articles_VrQRvur]: http://www.tuicool.com/articles/VrQRvur
[http_blog.csdn.net_lzyzsd_article_details_41833541]: http://blog.csdn.net/lzyzsd/article/details/41833541
[https_github.com_ReactiveX_RxJava]: https://github.com/ReactiveX/RxJava
[Link 1]: http://blog.csdn.net/EthanCo/article/details/49718775
[retrolambda]: https://github.com/evant/gradle-retrolambda
[android_JAVA8_retrolambda]: http://www.open-open.com/lib/view/open1433898197176.html
[Link 2]: http://download.csdn.net/detail/ethanco/9259239