RxJava初次使用2

1.订阅的小说更新了10章

2.代码

for(int i=0;i<10;i++){
                            Thread.sleep(1000);
//                        发送数据
                            e.onNext("更新一章"+i);
                        }
//                         关闭数据发送，无法再使用onNext();
                        e.onComplete();

使用for循环更新10章小说

RxJava初次使用2

3.同时订阅两本小说，同时更新10章

因为两本小说是不同的作者写的，所以小说的更新应当在不同的子线程中


io.reactivex.Observable<String> xiaoshuo1= io.reactivex.Observable.create(new ObservableOnSubscribe<String>() {
                            @Override
                            public void subscribe(ObservableEmitter<String> e) throws Exception {
                                for(int i=0;i<10;i++){
                                    Thread.sleep(1000);
//                        发送数据
                                    e.onNext("小说1"+"更新一章"+i);
                                }
//                         关闭数据发送，无法再使用onNext();
                                e.onComplete();
                    }
                }).subscribeOn(Schedulers.newThread());//将各个小说放到独立的子线程中

                io.reactivex.Observable<String> xiaoshuo2= io.reactivex.Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> e) throws Exception {
                        for(int i=0;i<10;i++){
                            Thread.sleep(1000);
//                        发送数据
                            e.onNext("小说2"+"更新一章"+i);
                        }
//                         关闭数据发送，无法再使用onNext();
                        e.onComplete();
                    }
                }).subscribeOn(Schedulers.newThread());


                io.reactivex.Observer<String> duzhe=new io.reactivex.Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        tv.setText("通知小说更新了");
                    }

                    @Override
                    public void onNext(String s) {
                        tv.setText(tv.getText()+"\n"+s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        tv.setText("错误");
                    }

                    @Override
                    public void onComplete() {
                        tv.setText(tv.getText()+"\n"+"开始阅读");
                    }
                };

//                读者订阅了小说
//                小说更新在子线程中，小说发生变化立刻通知读者
                io.reactivex.Observable
//                        merge同一接收小说更新通知
                        .merge(xiaoshuo1,xiaoshuo2)
                        .subscribeOn(Schedulers.newThread())
                        .observeOn(AndroidSchedulers.mainThread()).subscribe(duzhe);
            }
        });

读者订阅了小说
//                小说更新在子线程中，小说发生变化立刻通知读者
                io.reactivex.Observable
//                        merge同一接收小说更新通知
                        .merge(xiaoshuo1,xiaoshuo2)
                        .subscribeOn(Schedulers.newThread())
                        .observeOn(AndroidSchedulers.mainThread()).subscribe(duzhe);

使用merge()方法统一接收小说更新通知

RxJava初次使用2