RxJava 备注

RxJava是一个采用观察者模式的异步框架，本文给出几个基本的使用例子。

1、配置依赖：

    compile 'io.reactivex:rxjava:1.0.14'
    compile 'io.reactivex:rxandroid:1.0.1'

2、基础（发布、订阅）：

observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Step1");
                subscriber.onNext("Step2");
                subscriber.onNext("Step3");
                subscriber.onCompleted();
            }
        });

        subscriber = new Subscriber<String>() {

            @Override
            public void onStart(){
                System.out.println("onStart");
            }

            @Override
            public void onCompleted() {
                System.out.println("onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }
            @Override
            public void onNext(String s) {
                System.out.println("onNext"+s);
            }
        };

        FloatingActionButton fab = (FloatingActionButton) findViewById(R.id.fab);
        fab.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                observable.subscribe(subscriber);
            }
        });

3、采用action的写法

Action1<String> onNextAction = new Action1<String>() {

            @Override
            public void call(String s) {
                System.out.println("onNextAction" + s);
            }
        };


        Action1<Throwable> onErrorAction = new Action1<Throwable>() {

            @Override
            public void call(Throwable throwable) {
                System.out.println("onErrorAction");
            }
        };


        Action0 onCompletedAction = new Action0() {

            @Override
            public void call() {
                System.out.println("onCompletedAction");
            }
        };

        //from表示从一个数组中取参数
        Observable.just("Step1","Step2")
                .subscribe(onNextAction,onErrorAction,onCompletedAction);

4、Schedule接口

//Schedulers.immediate():默认值，当前线程
        //Schedulers.computation():用以CPU密集型任务，固定线程池
        //Schedulers.io():IO操作，不固定线程池
        //Schedulers.newThread()://新线程
        Observable.just("Step1","Step2")
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()) //主线程
                .subscribe(onNextAction, onErrorAction, onCompletedAction);

5、参数映射

List<String> input = new ArrayList<String>();
        input.add("hello ");
        input.add("world");

        Observable.just(input)
                .map(new Func1<List<String>, String>() {
                    @Override
                    public String call(List<String> input) {

                        StringBuffer buffer = new StringBuffer();

                        for (String item : input) {
                            buffer.append(item);
                        }

                        return buffer.toString();
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String str) {
                        System.out.println(str);
                    }
                });