记：RxJava小词典

一.使用Disposable切断被观察与观察者之间连接

语法：disposable.dispose();

public class SimpleActivity extends Activity {
    private static final String TAG = "SimpleActivity";
    private Disposable disposable;

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                //2
                Log.e(TAG, "subscribe");
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                //1
                Log.e(TAG, "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                //3
                Log.e(TAG, "onNext收到：" + integer);
                if (integer == 2) {
                    disposable.dispose();//切断被观察者与观察者之间连接
                    Log.e(TAG, "切断了连接");
                }
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.e(TAG, "onError");

            }

            @Override
            public void onComplete() {
                //4
                Log.e(TAG, "onComplete");

            }
        });
    }
}

打印为：
E/SimpleActivity: onSubscribe
E/SimpleActivity: subscribe
E/SimpleActivity: onNext收到：1
E/SimpleActivity: onNext收到：2
E/SimpleActivity: 切断了连接

---

二.被观察者的多种创建方式

1.just()
语法：
Observable.just(1, 2, 3)
注意事项：
（1）只能发送10个以内事件

        Observable.just(1, 2, 3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        //1
                        Log.e(TAG,"onSubscribe");
                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        //2
                        Log.e(TAG,"onNext收到："+integer);

                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        //3
                        Log.e(TAG,"onComplete");

                    }
                });

打印为：
E/CaoZuoFuActivity: onSubscribe
E/CaoZuoFuActivity: onNext收到：1
E/CaoZuoFuActivity: onNext收到：2
E/CaoZuoFuActivity: onNext收到：3
E/CaoZuoFuActivity: onComplete

2.fromArray()
语法：
Observable.fromArray(nums)
注意事项：
（1）可发送超过10个事件
（2）传输对象为数组

        Integer[] nums={1,2,3};
        Observable.fromArray(nums)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        //1
                        Log.e(TAG,"onSubscribe");
                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        //2
                        Log.e(TAG,"onNext收到："+integer);

                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        //3
                        Log.e(TAG,"onComplete");

                    }
                });

E/CaoZuoFuActivity: onSubscribe
E/CaoZuoFuActivity: onNext收到：1
E/CaoZuoFuActivity: onNext收到：2
E/CaoZuoFuActivity: onNext收到：3
E/CaoZuoFuActivity: onComplete

3.fromIterable()
语法：
Observable.fromIterable(list)
注意事项：
（1）可发送超过10个事件
（2）传输对象为集合

        List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        Observable.fromIterable(list)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        //1
                        Log.e(TAG,"onSubscribe");
                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        //2
                        Log.e(TAG,"onNext收到："+integer);

                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        //3
                        Log.e(TAG,"onComplete");

                    }
                });

E/CaoZuoFuActivity: onSubscribe
E/CaoZuoFuActivity: onNext收到：1
E/CaoZuoFuActivity: onNext收到：2
E/CaoZuoFuActivity: onNext收到：3
E/CaoZuoFuActivity: onComplete

---

三.延迟创建

1.defer()
语法：
Observable.defer(new Callable<ObservableSource<? extends Integer>)
注意事项：
（1）只有当被观察者注册时，才会创建被观察者

 Integer i = 5;
 Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> call() throws Exception {
                return Observable.just(i);
            }
        });

        i = 1;

        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Integer integer) {
                Log.e(TAG, "收到：" + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

收到：1

2.timer()
语法：
Observable.timer(2, TimeUnit.SECONDS).subscribe()
注意事项：
（1）延迟指定时间后，发送1个0。相当于延迟加载多少执行，执行某个事情
（2）timer操作符默认运行在一个新线程上

         Observable.timer(2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Long aLong) {
                Log.e(TAG, "接收到了事件"+ aLong  );

            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

E/TimeActivity: 接收到了事件0
（时间为2秒后）

---

三.操作符

1.map()
定义：
将发送的事件类型转换为指定的类型
语法：
map(new Function<转换前类型, 转换后类型>()
注意事项：
（1）map操作符中重写的apply方法，返回要写，不能使用默认的null

        Observable.create(new ObservableOnSubscribe<Integer>() {

            // 1. 被观察者发送事件 = 参数为整型 = 1、2、3
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);

            }
            // 2. 使用Map变换操作符中的Function函数对被观察者发送的事件进行统一变换：整型变换成字符串类型
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return integer+"转换后" ;
            }
        }).subscribe(new Consumer<String>() {

            // 3. 观察者接收事件时，是接收到变换后的事件 = 字符串类型
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, s);
            }
        });

E/MapActivity: 1转换后
E/MapActivity: 2转换后
E/MapActivity: 3转换后

2.flatMap()
定义：
将发送的事件进行拆分，每个拆分后的事件再单独转换，再合并成新的事件。所有新的事件再组合成新的事件，发送给观察者
语法：
flatMap(new Function<Integer, ObservableSource<String>>()
注意事项：
（1）新合并后的事件是无序的，不与拆分前顺序一致

flatMap流程.jpg

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                //先拆分，然后转换为String，然后组合成集合
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("我是事件 " + integer + "拆分后的子事件" + i);
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, s);
            }
        });

E/MapActivity: 我是事件 1拆分后的子事件0
E/MapActivity: 我是事件 1拆分后的子事件1
E/MapActivity: 我是事件 1拆分后的子事件2
E/MapActivity: 我是事件 2拆分后的子事件0
E/MapActivity: 我是事件 2拆分后的子事件1
E/MapActivity: 我是事件 2拆分后的子事件2
E/MapActivity: 我是事件 3拆分后的子事件0
E/MapActivity: 我是事件 3拆分后的子事件1
E/MapActivity: 我是事件 3拆分后的子事件2

2.ConcatMap()
用法和概念和flatMap()类似，区别在于concatMap是重新合并后的顺序与原来顺序一致。