2017 September 25 Java

Java-RxJava2笔记

参考：
http://blog.csdn.net/maplejaw_/article/details/52442065
http://www.jianshu.com/nb/5864063

去年RxJava2.x发布了，与RxJava1.x相比，使用上有不少改动(只是API函数名改了,而使用流程思维不变)，故在此记录笔记存档！

一.基本用法

1.创建Observable(被观察者/发布者/发射者)

(1)create()
Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
        observableEmitter.onNext("发布数据1");
        observableEmitter.onNext("发布数据2");
        observableEmitter.onError(new Throwable("发生错误"));          
        observableEmitter.onComplete(); //完成
    }
});

(2)just
Observable observable = Observable.just("发布数据1", "发布数据2");

(3)fromIterable, fromArray
ArrayList<String> list = new ArrayList<>();
list.add("发布数据1");
list.add("发布数据2");
Observable observable = Observable.fromIterable(list);

(4)range,第一个参数为起始值,第二个为发送的个数,如果为0则不发送，负数则抛异常
Observable observable = Observable.range(10, 5)

(5)defer,延期,有观察者订阅时才创建Observable
Observable observable = Observable.defer(new Callable<ObservableSource>() {
    @Override
    public ObservableSource call() throws Exception {
        return Observable.just("发布数据1","发布数据2");
    }
});

(6)interval,定时周期发布数据
 Observable observable = Observable.interval(500, TimeUnit.MILLISECONDS); //周期500ms

(7)timer,延迟发布数据
Observable observable = Observable.timer(300, TimeUnit.MILLISECONDS); //延迟300ms

(8)repeat,重复发布数据
Observable observable = Observable.just("发布数据1").repeat(3); //重复发布3次

2.创建Observer(观察者/订阅者/接收者)

(1).Observer完整形式
Observer observer = new Observer() {
    @Override
    public void onSubscribe(@NonNull Disposable disposable) {
        //Disposable 相当于RxJava1.x中的Subscription,用于解除订阅
    }

    @Override
    public void onNext(@NonNull Object o) {
        //接收数据
    }

    @Override
    public void onError(@NonNull Throwable throwable) {
        //接收错误
    }

    @Override
    public void onComplete() {
        //通知完成
    }
};

(2).Observer简写形式
Consumer onNext = new Consumer() {//接收数据
    @Override
    public void accept(Object o) throws Exception {
    }
};

Consumer<Throwable> onError = new Consumer<Throwable>() {//接收错误
    @Override
    public void accept(Throwable throwable) throws Exception {
    }
};

Action onComplete = new Action() {//通知完成
    @Override
    public void run() throws Exception {
    }
};

Consumer<Disposable> onSubscribe = new Consumer<Disposable>() {
    @Override
    public void accept(Disposable disposable) throws Exception {
    }
};

3.Observer订阅Observable

(1).Observer完整订阅
observable.subscribe(observer); //订阅

(2).Observer简写订阅
observable.subscribe(onNext);
observable.subscribe(onNext, onError);
observable.subscribe(onNext, onError, onComplete);
observable.subscribe(onNext, onError, onComplete, onSubscribe);

二.线程调度

调度器类型
Schedulers.computation( )  用于计算任务，如事件循环或和回调处理，不要用于IO操作(IO操作请使用Schedulers.io())；默认线程数等于处理器的数量
Schedulers.from(executor)  使用指定的Executor作为调度器
Schedulers.io( )	       用于IO密集型任务，如异步阻塞IO操作，这个调度器的线程池会根据需要增长；
                           对于普通的计算任务，请使用Schedulers.computation()；
                           Schedulers.io( )默认是一个CachedThreadScheduler，很像一个有线程缓存的新线程调度器
Schedulers.newThread( )    为每个任务创建一个新线程
Schedulers.trampoline( )   当其它排队的任务完成后，在当前线程排队开始执行
AndroidSchedulers.mainThread()  此调度器为RxAndroid特有，顾名思义，运行在Android UI线程上

Observable.just("耗时操作...")
    .subscribeOn(Schedulers.io())//io线程-发布者
    .observeOn(AndroidSchedulers.mainThread())//主线程-接收者
    .subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {                        
        }
    });

三.常用操作符

1.map-数据类型转换

Observable.just("123")
    .map(new Function<String, Integer>() {
        @Override
        public Integer apply(@NonNull String s) throws Exception {
            return Integer.parseInt(s);
        }
    })
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

2.flatMap-数据集合扁平化(遍历循环每一个元素)

List<String> list = new ArrayList<>();
list.add("1");
list.add("2");
list.add("3");
List<List<String>> listSSS = new ArrayList<>();//二维数组集合
listSSS.add(list);
Observable.fromIterable(listSSS)
.flatMap(new Function<List<String>, ObservableSource<String>>() {
    @Override
    public ObservableSource<String> apply(@NonNull List<String> list) throws Exception {
        return Observable.fromIterable(list);
    }
})
.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});

3.buffer-缓存满后,以list集合发送数据

List<String> list = new ArrayList<>();
list.add("1");
list.add("2");
list.add("3");
Observable.fromIterable(list)
    .buffer(list.size())  //缓存一起发送
    .subscribe(new Consumer<List<String>>() {
        @Override
        public void accept(List<String> list) throws Exception {
            System.out.println(list.size());
        }
    });

4.take(n)-发送前n项数据

Observable.just(1, 2, 1, 1, 2, 3)
    .take(3) //发送前3项数据
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

5.distinct-去除重复项

Observable.just(1, 2, 1, 1, 2, 3)
    .distinct() //去重
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

6.filter-过滤

Observable.just(1, 2, 3, 4, 5)
    .filter(new Predicate<Integer>() {
        @Override
        public boolean test(@NonNull Integer integer) throws Exception {
            return integer > 3; //过滤大于3
        }
    })
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
        }
    });

五.Flowable-背压

Flowable是RxJava2.x中新增的类，专门用于应对背压Backpressure问题
背压: 即生产者的速度大于消费者的速度带来的问题，比如在Android中常见的点击事件，点击过快则会造成点击两次的效果!
Flowable.create(new FlowableOnSubscribe<Integer>() {
        @Override
        public void subscribe(FlowableEmitter<Integer> e) throws Exception {
            for (int i = 0; i < 10000; i++)
                e.onNext(i);
            e.onComplete();
        }
}, BackpressureStrategy.ERROR) //指定背压处理策略，抛出异常错
    .subscribeOn(Schedulers.computation())
    .observeOn(Schedulers.newThread())
    .subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println(integer);
            Thread.sleep(1000);
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            System.out.println(throwable);
        }
    });

// 如Rxjava1.x一样简写
Flowable.range(1,10000)
.onBackpressureDrop() // 背压
.subscribe(new Consumer<Integer>() {
    @Override
    public void accept(Integer integer) throws Exception {
        System.out.println(integer);
    }
});

四.Subject

Subject extends Observable implements Observe
作用:
    可充当Observable
    可充当Observer
    是Observable和Observer之间的桥梁        
Subject有四个实现类: AsyncSubject, BehaviorSubject, PublishSubject, ReplaySubject
注意：
    从多个线程中调用onNext(on系列方法),需要使用串行化Serialized,才能顺序调用！
    SerializedSubject<String, Integer> ser = new SerializedSubject(publishSubject);

Processor和Subject的作用相同,其中Processor是RxJava2.x新增的,继承自Flowable,所以支持背压控制
//Processor
AsyncProcessor<String> processor = AsyncProcessor.create();
processor.subscribe(o -> Log.d("JG",o)); //three
processor.onNext("one");
processor.onNext("two");
processor.onNext("three");
processor.onComplete();

1.AsyncSubject只接收onCompleted()被调用前的最后一个数据

AsyncSubject<String> asyncSubject = AsyncSubject.create();
asyncSubject.onNext("asyncSubject1");
asyncSubject.onNext("asyncSubject2");
asyncSubject.onNext("asyncSubject3");
asyncSubject.onComplete();
asyncSubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);//只接收到asyncSubject3
    }
});    

2.BehaviorSubject接收被订阅前的最后一个数据,还接收订阅后的数据

BehaviorSubject<String> behaviorSubject = BehaviorSubject.create();
behaviorSubject.onNext("behaviorSubject1");
behaviorSubject.onNext("behaviorSubject2");
behaviorSubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s); //接收到behaviorSubject2, behaviorSubject3, behaviorSubject4
    }
});
behaviorSubject.onNext("behaviorSubject3");
behaviorSubject.onNext("behaviorSubject4");

3.PublishSubject只接收被订阅后的数据

PublishSubject<String> publishSubject = PublishSubject.create();
publishSubject.onNext("publishSubject1");
publishSubject.onNext("publishSubject2");
publishSubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);只接收到behaviorSubject3, behaviorSubject4
    }
});
publishSubject.onNext("publishSubject3");
publishSubject.onNext("publishSubject4");

4.ReplaySubject接收所有数据,无论何时订阅! 但缓存到一定大小时或一段时间后会丢弃旧的数据！

ReplaySubject<String> replaySubject = ReplaySubject.create(); //默认初始缓存容量大小为16
//replaySubject = ReplaySubject.create(100);//指定初始缓存容量大小为100
//replaySubject = ReplaySubject.createWithSize(2);//只缓存订阅前最后2条数据
//replaySubject = ReplaySubject.createWithTime(1,TimeUnit.SECONDS,Schedulers.computation());//只缓存被订阅前1秒内的数据
replaySubject.onNext("replaySubject:pre1");
replaySubject.onNext("replaySubject:pre2");
replaySubject.onNext("replaySubject:pre3");
replaySubject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});
replaySubject.onNext("replaySubject:after1");
replaySubject.onNext("replaySubject:after2");

5.Subject作为桥梁,使用示例

//1.Subject作为桥梁
Subject<String> subject = BehaviorSubject.create();

//2.订阅
subject.subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});

//3.发布数据
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> observableEmitter) throws Exception {
        observableEmitter.onNext("as Bridge");
    }
}).subscribe(subject);

简书: http://www.jianshu.com/p/724c937e3d0c
CSDN博客: http://blog.csdn.net/qq_32115439/article/details/78090944
GitHub博客: http://lioil.win/2017/09/25/JavaSE-RxJava.html
Coding博客: http://c.lioil.win/2017/09/25/JavaSE-RxJava.html