Android 带大家进入RxJava的世界

2016-09-23 本文已影响855人肖赛Soaic

Rx指的是ReactiveX，也是Reactive Extensions的缩写，是一个使用可观察数据流进行异步编程的编程接口，结合了观察者模式、迭代器模式和函数式编程的精华，是一种编程思想的突破，它影响了许多其它的程序库和框架以及编程语言

RxJava里最核心的两个东西Observable和Subscriber，Observable指的是被观察者、事件源，Subscriber指的是被观察者、订阅者、用户。

举个简单的例子，像微信里面的服务号和公众号，我们关注了某个服务号或公众号，只要发出消息，我们都能同时收到。这里的服务号和公众号就是Observable，Subscriber指的就是我们用户。

看起来RxJava蛮像观察者模式，不过有一点不同Observable如果没有Subscriber不会发出任何事件

下面就让我们一起进入RxJava的世界吧

参考资料学习网址

1、Grokking RxJava
2、github地址
3、ReactiveX/RxJava文档中文版
4、ReactiveX官网
5、抛物线博客

配置环境

在builde.gradle里面添加

compile 'io.reactivex:rxandroid:1.2.1'
compile 'io.reactivex:rxjava:1.1.6'

基本实现

1、Hello world

创建Observable对象，通过Observable.create()来创建，sub.onNext()可以发出信息，最后调用sub.onCompleted()来完成，还有一个sub.onError()方法是出现异常提供的一个方法，一旦调用了sub.onError()或者sub.onCompleted()，后面的逻辑代码都不会执行。

Observable<String> myObservable = Observable.create(
        new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> sub) {
                sub.onNext("Hello, world1!");
                sub.onNext("Hello, world2!");
                sub.onCompleted();
            }
        }
);

创建Subscriber对象，实例化Subscriber抽象类，实现onNext、onCompleted、onError三个方法，依次响应Observable对象里面的sub.onNext()、sub.onCompleted()和sub.onError()方法

Subscriber<String> mySubscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d("rxjava","onNext="+s); 
    }
    @Override
    public void onCompleted() { 
        Log.d("rxjava","onCompleted");
    }
    @Override
    public void onError(Throwable e) {
        Log.d("rxjava","onError="+e);
    }
};

最后Subscriber订阅Observable，这里Observable可以被多个Subscriber订阅

myObservable.subscribe(mySubscriber);
myObservable.subscribe(mySubscriber2);
myObservable.subscribe(mySubscriber3);

2、简化代码

上面的代码可以转变为这样：

Observable<String> myObservable = Observable.just("Hello, world1!"，"Hello, world2!");
Action1<String> onNextAction = new Action1<String>() {
    @Override
    public void call(String s) {
        System.out.println(s);
    }
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
    @Override
    public void call(Throwable e) {
    }
};
Action0 onCompletedAction = new Action0() {
    @Override
    public void call() {
    }
};
// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
myObservable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
myObservable.subscribe(onNextAction,onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
myObservable.subscribe(onNextAction,onErrorAction,onCompletedAction);

Observable.just("Hello, world1!"，"Hello, world2!")会直接创建Observable对象，
然后依次调用sub.onNext("Hello, world1!"); sub.onNext("Hello, world2!"); sub.onCompleted();

一般onErrorAction和onCompletedAction我们可以不用管理最后代码可优化成：

Observable.just("Hello, world1!","Hello, world2!")
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        System.out.println(s);
                    }
                });

java8语法lambda可以写成

Observable.just("Hello, world!")
                .subscribe(s -> System.out.println(s));

3、操作符map

简单的一个例子，传递一个字符串，然后获取它的长度，最后打印出来，在RxJava里我们可以这样实现：

Observable.just("Hello, world!")
        .map(new Func1<String,Integer>(){
            @Override
            public Integer call(String s){
                return s.length();
            }
        })
        .subscribe(new Action1<Integer>(){
            @Override
            public void call(Integer i){
                System.out.println(Integer.toString(i));
            }
        });

这里map方法需要结合Func1接口来使用，Func1里第一个String类型是对应接收just类型，第二个Integer类型是返回值，供下一个对象接收

4、操作符from

如果我们要循环遍历一个数组或者集合，在RxJava里面可以使用from操作符来实现

String[] items = { "0", "1", "2", "3", "4", "5" };
        Observable.from(items)
                .subscribe(new Action1<String>(){
                    @Override
                    public void call(String str){
                        //依次遍历打印items
                        System.out.println(str);
                    }
                });

操作符from会循环遍历方法参数里面的数组或者集合，然后依次调用onNext()方法

5、操作符flatMap

如果我们要实现多重循环，可以使用flatMap操作符来实现

String[][] itemArray = {{"1","2","3"},{"4","5","6"},{"7","8","9"}};
Observable.from(itemArray)
        .flatMap(new Func1<String[],Observable<String>>(){
            @Override
            public Observable<String> call(String[] s){
                return Observable.from(s);
            }
        })
        .subscribe(new Subscriber<String>(){
            @Override
            public void onCompleted(){
            }
            @Override
            public void onError(Throwable e){
            }
            @Override
            public void onNext(String str){
                Log.d("rxJava","str="+str);
            }
        });

上面代码执行后，依次打印str=1到9，这里要注意的是Func1第二个参数类型是Observable<?>

6、操作符filter

字面理解应该就知道它的作用了，过滤不需要的数据

Observable.just(1, 2, 3, 4, 5)
    .filter(new Func1<Integer, Boolean>() {
        @Override
        public Boolean call(Integer item) {
            return( item < 4 );
        }
    })
    .subscribe(new Subscriber<Integer>() {
        @Override
        public void onNext(Integer item) {
            System.out.println("Next: " + item);
        }
        @Override
        public void onError(Throwable error) {
            System.err.println("Error: " + error.getMessage());
        }
        @Override
        public void onCompleted() {
            System.out.println("Sequence complete.");
        }
    });

上面代码执行结果，只会打印1到3，后面4和5不符合，则过滤

7、操作符merge

合并多个Observables

Observable<Integer> odds = Observable.just(1, 3, 5);
Observable<Integer> evens = Observable.just(2, 4, 6);
Observable.merge(odds, evens)
        .subscribe(new Subscriber<Integer>() {
            @Override
            public void onNext(Integer item) {
                System.out.println("Next: " + item);
            }
            @Override
            public void onError(Throwable error) {
                System.err.println("Error: " + error.getMessage());
            }
            @Override
            public void onCompleted() {
                System.out.println("Sequence complete.");
            }
        });

上面代码执行后结果：依次打印的是1、3、5、2、4、6

*  Javadoc: merge(Iterable)
*  Javadoc: merge(Iterable,int)
*  Javadoc: merge(Observable[])
*  Javadoc: merge(Observable,Observable) (接受二到九个Observable)

除了传递多个Observable给merge，你还可以传递一个Observable列表List，数组，甚至是一个发射Observable序列的Observable，merge将合并它们的输出作为单个Observable的输出

8、操作符timer和interval

5秒后打印数据，第一个参数是多少时间执行，第二个参数是时间单位

Observable.timer(5,TimeUnit.SECONDS)
        .subscribe(new Action1<Long>(){
            @Override
            public void call(Long aLong){
                Log.d("rxJava","5秒后执行"+aLong);
            }
        });

1秒后每5秒循环打印，第一个参数为多少时间后执行，第二个为没多少时间执行，第三个是时间单位

Observable.interval(1,5,TimeUnit.SECONDS)
        .subscribe(new Action1<Long>(){
            @Override
            public void call(Long aLong){
                Log.d("rxJava","1秒后每5秒循环执行"+aLong);
            }
        });

这里aLong返回的值应该是打印的次数

9、线程切换

主要通过subscribeOn()和 observeOn()来实现
subscribeOn()是控制Observable.OnSubscribe的线程切换， subscribeOn()的线程切换发生在 OnSubscribe中，即在它通知上一级 OnSubscribe时，这时事件还没有开始发送，因此控制可以从事件发出的开端就造成影响
observeOn()则是控制Subscriber的线程切换，observeOn() 的线程切换则发生在它内建的 Subscriber中，即发生在它即将给下一级 Subscriber发送事件时，因此控制的是它后面的线程
Schedulers.io()：IO线程
AndroidSchedulers.mainThread()：主线程
Schedulers.newThread()：新开线程

Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread())
    .map(mapOperator) // 新线程，由 observeOn() 指定
    .observeOn(Schedulers.io())
    .map(mapOperator2) // IO 线程，由 observeOn() 指定
    .observeOn(AndroidSchedulers.mainThread) 
    .subscribe(subscriber);  // Android 主线程，由 observeOn() 指定

10、防止按钮重复点击

1秒钟内只会执行第一次点击

RxView.clicks(view)
        .throttleFirst(1,TimeUnit.SECONDS)
        .subscribe(new Observer<Object>(){
            @Override
            public void onCompleted(){
            }
            @Override
            public void onError(Throwable e){
            }
            @Override
            public void onNext(Object o){
            }
        });

RxView是RxBinding库里的对象，需要添加环境

compile 'com.jakewharton.rxbinding:rxbinding:0.4.0'

throttleFirst():在每次事件触发后的一定时间间隔内丢弃新的事件

11、RxBus代替EventBus

public class RxBus {
    // 主题
    private final Subject<Object, Object> bus;
    // PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
    private RxBus() {
        bus = new SerializedSubject<>(PublishSubject.create());
    }
    public static RxBus get() {
        return RxBusHolder.sInstance;
    }
    public void unSubscribe(CompositeSubscription compositeSubscription){
        if (compositeSubscription != null && !compositeSubscription.isUnsubscribed())
            compositeSubscription.unsubscribe();
    }
    private static class RxBusHolder {
        private static final RxBus sInstance = new RxBus();
    }
    // 提供了一个新的事件
    public void post(Object o) {
        bus.onNext(o);
    }
    // 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
    public <T> Observable<T> toObserverable(Class<T> eventType) {
        return bus.ofType(eventType);
    }
}

RxBus提供了三个方法toObserverable()获取Observable对象、unSubscribe()解除订阅、post()发送数据。

在MainActivity中实现

    public class MainActivity extends AppCompatActivity {
        private CompositeSubscription allSubscription = new CompositeSubscription();
        @Override
        protected void onCreate(Bundle savedInstanceState){
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_main);
            ...
            initRxBus();
        }
        private void initRxBus(){
            //注册订阅RxBus并添加到CompositeSubscription,在onDestroy()里可以统一解除订阅
            allSubscription.add(RxBus.get().toObserverable(OneEvent.class).subscribe(this::response));
        }
        @Override
        protected void onDestroy(){
            super.onDestroy();
            //解除订阅
            RxBus.get().unSubscribe(allSubscription);
        }
        /**
         * RxBus响应接收数据方法
         * @param event
         */
        public void response(OneEvent event) {
            Toast.makeText(getApplicationContext(),event.msg,Toast.LENGTH_LONG).show();
        }
        class OneEvent {
            String msg;
            public OneEvent(String msg) {
                this.msg = msg;
            }
        }
    }

最后在需要发送数据的地方调用，跟EventBus还是满相似的

RxBus.get().post(new OneEvent("hello bus"));

12、结合Retrofit使用
这里我就直接拿介绍Retrofit那篇文章的例子在这里说明下

@POST("query")
Observable<PostQueryInfo> searchRx(@Query("type") String type, @Query("postid") String postid);

        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://www.kuaidi100.com/")
                 //添加数据解析ConverterFactory
                .addConverterFactory(GsonConverterFactory.create()) 
                 //添加RxJava
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())   
                .build();
        GitHubService apiService = retrofit.create(GitHubService.class);
        apiService.searchRx("yuantong","500379523313")
                //访问网络切换异步线程
                .subscribeOn(Schedulers.io())
                //响应结果处理切换成主线程
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<PostQueryInfo>() {
                    @Override
                    public void onCompleted() {
                         //请求结束回调
                    }
                    @Override
                    public void onError(Throwable e) {
                         //错误回调
                        e.printStackTrace();
                    }
                    @Override
                    public void onNext(PostQueryInfo postQueryInfo) {
                         //成功结果返回
                        Log.e("APP",postQueryInfo.getNu());
                    }
                });

Retrofit支持RxJava可以返回Observable对象，所以我们就可以直接拿来用，很方便吧
.subscribeOn(Schedulers.io()请求网络切换成IO线程
.observeOn(AndroidSchedulers.mainThread())切换成主线程更新UI
onNext(PostQueryInfo postQueryInfo)成功回调解析数据
public void onError(Throwable e)异常回调抛出异常

Tips

配置java8 lambda环境

1、在Module中的build.gradle中添加，然后Make Module 'xxx'

apply plugin: 'me.tatarka.retrolambda'
buildscript {
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath 'me.tatarka:gradle-retrolambda:3.2.5'
    }
}
repositories {
    mavenCentral()
}
android {
    compileOptions {
        sourceCompatibility JavaVersion.VERSION_1_8
        targetCompatibility JavaVersion.VERSION_1_8
    }
    ...
}

2、测试下面代码不报红，配置环境OK

textView.setOnClickListener(v -> 
    Toast.makeText(getApplicationContext(),"lambda",Toast.LENGTH_LONG).show()
);

学到最后其实RxJava只是刚入了个门，还有很多很多需要我们去学习的，这里只是跟大家一起看了看RxJava的世界是什么样子的，如果要学好，还是要多去看看api文档，多练习才能真正的熟练使用它。

革命尚未成功，我们仍需努力

Android 带大家进入RxJava的世界

参考资料学习网址

配置环境

基本实现

Tips

猜你喜欢

热点阅读