标签：Observable onNext System println 使用手册 Rxjava emitter out

创建Observables

https://github.com/ReactiveX/RxJava/wiki/Creating-Observables#create

1：Observable.fromAction使用效果：

Observable<Integer> observable = Observable.fromAction(
                () -> {
                    System.out.println("我去，被执行了");
                }
        );
        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("onSubscribe1");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("onNext1" + integer);
            }

            @Override
            public void one rror(@NonNull Throwable e) {
                System.out.println("onError1");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete1");
            }
        });
        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("onSubscribe2");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("onNext2" + integer);
            }

            @Override
            public void one rror(@NonNull Throwable e) {
                System.out.println("onError2");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete2");
            }
        });

onSubscribe1
我去，被执行了
onComplete1
onSubscribe2
我去，被执行了
onComplete2

 

2：Observable.just(有限个item)

内部使用fromArray()

 

3：Observable.create()

Observable<String> observable1 = Observable.create((emitter -> {
            emitter.onNext("a");
            emitter.onNext("b");
            emitter.onNext("c");
            emitter.onNext("d");
            emitter.onComplete();这边调用onNext等方法
        }));

        observable1.subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println("onNext");
            }

            @Override
            public void one rror(@NonNull Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });

 

官方例子：

ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

        ObservableOnSubscribe<String> handler = emitter -> {

            Future<Object> future = executor.schedule(() -> {
                emitter.onNext("Hello");
                emitter.onNext("World");
                return "result";
            }, 3, TimeUnit.SECONDS);

            String result = (String) future.get();//阻塞
            emitter.onNext(result);
            emitter.onComplete();//onComplete之后调用onNext无用
            emitter.setCancellable(() -> future.cancel(false));
        };

        Observable<String> observable = Observable.create(handler);

        observable.subscribe(item -> System.out.println(item), error -> error.printStackTrace(),
                () -> System.out.println("Done"));
        
        executor.shutdown();

 注：需要注意回调的调用时机

 Observable.create(ObservableOnSubscribe<Int> { emitter ->
                    log("create ${Thread.currentThread().name}")
                    emitter.onNext(1)
                    emitter.onNext(2)
                    emitter.onNext(3)
                    emitter.onComplete()
                }).compose(RxUtil.ioAndMainObservable())
                    .subscribe(object : Observer<Int> {
                        override fun onSubscribe(d: Disposable) {
                          log("onSubscribe ${Thread.currentThread().name}")
                        }

                        override fun onNext(t: Int) {
                            log("onNext ${Thread.currentThread().name}")
                        }

                        override fun one rror(e: Throwable) {
                            log("onError ${Thread.currentThread().name}")
                        }

                        override fun onComplete() {
                            log("onComplete ${Thread.currentThread().name}")
                        }
                    })

onSubscribe main
create RxCachedThreadScheduler-1
onNext main
onNext main
onComplete main

就是说，onScubcribe在emitter的那段代码之前被调用，并且subscribeOn切换的线程只是emitter中的，不影响onSubscribe，onSubscribe的线程和onNext一样。

 

4：处理异常

官方例子：https://github.com/ReactiveX/RxJava/wiki/How-To-Use-RxJava#error-handling

 

Hot & Cold Observable

When does an Observable begin emitting its sequence of items? It depends on the Observable. A “hot” Observable may begin emitting items as soon as it is created, and so any observer who later subscribes to that Observable may start observing the sequence somewhere in the middle. A “cold” Observable, on the other hand, waits until an observer subscribes to it before it begins to emit items, and so such an observer is guaranteed to see the whole sequence from the beginning.

In some implementations of ReactiveX, there is also something called a “Connectable” Observable. Such an Observable does not begin emitting items until its Connect method is called, whether or not any observers have subscribed to it.

 

 

Scheduler

https://reactivex.io/documentation/scheduler.html

 

 

 

巨人的肩膀：

一些很重要的概念：https://reactivex.io/documentation/observable.html

标签：Observable,onNext,System,println,使用手册,Rxjava,emitter,out
来源： https://www.cnblogs.com/--here--gold--you--want/p/15811660.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。