RxJava操作符记录

• Rxjava, 官方介绍：RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM. 中文：
  RxJava - JVM的Reactive Extensions - 一个使用Java VM的可观察序列组成异步和基于事件的程序的库。
• Filter()
  作用： 过滤特定条件的事件
• 原理
  Filter.png
• 代码示例

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // 1. 发送5个事件
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
                emitter.onNext(5);
            }

            // 2. 采用filter（）变换操作符
        }).filter(new Predicate<Integer>() {
            // 根据test()的返回值 对被观察者发送的事件进行过滤 & 筛选
              // a. 返回true，则继续发送
              // b. 返回false，则不发送（即过滤）
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer > 3;
                // 本例子 = 过滤了整数≤3的事件
            }
        }).subscribe(new Observer<Integer>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "过滤后得到的事件是："+ value  );
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        });

• flatMap()
  -将观察者发送的事件序列进行拆分&或单独转换，合并成一个新的实践序列进行发送
  -原理：
  1.为时间序列中的每个事件都创建一个Observable;
  2.将对每个 原始事件转换后的新事件 都放到对应的Observable;
  3.将每个新建的Observable放入到一个新建的、总的Observable对象;
  4.新建的、总的Observable 将合并的事件序列发送给观察者（Observer）；
  FlatMap.png

-代码示例：

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list=new ArrayList<>();
                list.add("我是事件 " + integer + "拆分后的子事件");
                return Observable.fromIterable(list);
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "accept: "+s );
                    }
                });

-打印出：

image.png