RxJava2简单使用二(Consumer)

时间:2022-02-19 17:47:12
        /**
         * Consumer是简易版的Observer,他有多重重载,可以自定义你需要处理的信息,我这里调用的是只接受onNext消息的方法,
         * 他只提供一个回调接口accept,由于没有onError和onCompete,无法再 接受到onError或者onCompete之后,实现函数回调。
         * 无法回调,并不代表不接收,他还是会接收到onCompete和onError之后做出默认操作,也就是监听者(Consumer)不在接收
         * Observable发送的消息,下方的代码测试了该效果。
         */
        final Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.d("MainActivity", Thread.currentThread().getName() + " String:" + s);
            }
        };

        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                Log.d("MainActivity", Thread.currentThread().getName() + "emit Hello");
                e.onNext("Hello");
                Log.d("MainActivity", Thread.currentThread().getName() + "emit Complete");
                e.onComplete();
                Log.d("MainActivity", Thread.currentThread().getName() + "emit World");
                e.onNext("World");
            }
        });