RxJava极简学习（四个简单例子）

RxJava

---

提到Rxjava最多人都是用来处理，线程调度，回调地狱，加上Retrofit又支持Rxjava，所以大部分开发者都只会在请求网络和需要切换线程的时候用到Rxjava，其实它有一个最重要的特性，它可以让数据的流向更加直观，代码更清晰。

举个栗子

---

比如说一个庞大的项目，一个事件传递的整个过程可能要经历很多方法，方法套方法，每个方法的位置七零八落，一个个方法跳进去看,跳过去跳过来很容易把脑袋弄晕，不够直观。但是Rxjava可以把所有逻辑用链式加闭包的方式呈现，做了哪些操作，谁在前谁在后非常直观，逻辑清晰，维护就会非常轻松。就算不是你写的你也可以很快的了解，你可以把它看作一条河流，整个过程就是对里面的水流做进行加工。懂了这个特性我们才知道在复杂的逻辑中运用Rxjava是多么的重要。

Gradle添加依赖

---

要在Android中使用RxJava2, 先添加Gradle配置。

compile "io.reactivex.rxjava2:rxjava:2.0.7"
compile "io.reactivex.rxjava2:rxandroid:2.0.1"

一个简单实例（分别创建被观察者，观察者和两种对象进行关联）

---

  //创建被观察者
 Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                //调用观察者的回调
                emitter.onNext("我是");
                emitter.onNext("RxJava");
                emitter.onNext("简单示例");
                emitter.onError(new Throwable("出错了"));
                emitter.onComplete();
            }
        });

        //创建观察者
        Observer<String> observer = new Observer<String>() {

            @Override
            public void onError(Throwable e) {
                Log.e(TAG,e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.e(TAG,"onCompleted");
            }

            //onSubscribe()方法是最先调用的
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG,"subscribe");
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG,s);
            }
        };

        //注册，将观察者和被观察者关联，将会触发OnSubscribe.call方法
        observable.subscribe(observer);

运行结果：

Observable是被观察者，创建后传入一个OnSubscribe对象，当Observable（观察者）调用subscribe进行注册观察者时，OnSubscribe的call方法会触发。
ObservableEmitter： Emitter 是发射器的意思，它可以发出三种类型的事件，与之对应的。
Observer有三个回调方法：

• onNext：接受到一个事件
• onCompleted：接受完事件后调用，只会调用一次
• onError ：发生错误时调用，并停止接受事件，调用一次

注：onCompleted和onError不会同时调用，只会调用其中之一

另一个简单实例（访问网络时，通过RxJava来进行快捷的主线程和子线程切换）

---

protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        final TextView tv_return = (TextView) findViewById(R.id.tv_return);
        //创建被观察者
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                emitter.onNext(getResponse());
            }
        });

        //创建观察者
        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String mResponse) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                tv_return.setText(mResponse);
            }
        };
       //subscribeOn() 指定的是发送事件的线程, observeOn() 指定的是接收事件的线程.
        observable.subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);

    }
    //使用okhttp访问网上提供的接口，由于是同步get请求，需要在子线程进行
    private String getResponse() {
        String url = "http://v.juhe.cn/weather/index?cityname=%E6%9D%AD%E5%B7%9E&dtype=&format=&key=7970495dbf33839562c9d496156e13cc";
        OkHttpClient client = new OkHttpClient();
        Request request = new Request.Builder()
                .url(url)
                .build();
        Response response;

        try {
            response = client.newCall(request).execute();
            return response.body().string();
        } catch (IOException e) {
            return "error";
        }
    }

在RxJava中, 已经内置了很多线程选项供我们选择, 例如有

• Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
• Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
• Schedulers.newThread() 代表一个常规的新线程
• AndroidSchedulers.mainThread() 代表Android的主线程
  这些内置的Scheduler已经足够满足我们开发的需求， 因此我们应该使用内置的这些选项，在RxJava内部使用的是线程池来维护这些线程，所有效率也比较高。

第三个简单实例（类型变换操作）

---

RxJava提供了类型变换操作，像上面的发送网络请求事件，得到网络请求数据，传递和接收的都是字符串。
而RxJava中的map操作符可以对事件进行处理，比如发送网络请求事件，通过map处理后，可以接收到网络请求的字符串的字节数，如下所示。

protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        final TextView tv_return = (TextView) findViewById(R.id.tv_return);
        //创建被观察者
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                emitter.onNext(getResponse());
            }
        //通过map操作符对数据进行中间处理
        }).map(new Function<String, Integer>() {
            @Override
            public Integer apply(@NonNull String response) throws Exception {
                return response.length();
            }
        });
        //创建观察者
        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                tv_return.setText("字数："+integer);
            }
        };
        //绑定，指定线程
        observable.subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);

    }

第四个简单实例（链式操作）

---

而RxJava一个引以为豪的地方就是它的链式操作了，可以把被观察者和观察者串起来。

protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        final TextView tv_return = (TextView) findViewById(R.id.tv_return);
        
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                emitter.onNext(getResponse());
            }
        }).map(new Function<String, Integer>() {
            @Override
            public Integer apply(@NonNull String response) throws Exception {
                return response.length();
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
                tv_return.setText("字数：" + integer);
            }
        });
    }

这里是项目地址 其实也没啥东西

喜欢的同学可以点个赞

参考

https://xiaobailong24.me/2017/03/18/Android-RxJava2.x/
http://www.jianshu.com/p/6fd8640046f1