rxJava学习笔记之subscribeOn指定线程

末蓝、 2022-07-14 04:08 242阅读 0赞

先上代码

longOpeSubscription = Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
                    printLog("onStart in OnSubscribe");
                    subscriber.onStart();
                    int N = data.length;
                    for (int i = 0; i < N; i++) {
                        dosomethingBlockThread();
                        printLog("onNext" + data[i] + " in OnSubscribe");
                        subscriber.onNext(data[i]);
    
                    }
                    printLog("OnCompleted in OnSubscribe");
                    subscriber.onCompleted();
                }
            }).subscribeOn(Schedulers.io())
                    .subscribe(new Subscriber<String>() {
                        @Override
                        public void onCompleted() {
                            printLog("OnCompleted in Subscriber");
    
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            printLog("onError in Subscriber");
    
                        }
    
                        @Override
                        public void onNext(String s) {
                            printLog("onNext " + s + " in Subscriber");
    
                        }
                    });
    ... java'''

首先我们知道create的时候创建了一个新的observable,和一个onSubscribe对象。接着subscribeOn的时候再新建了一个observable,和一个onSubscribe对象。所以最后subscribe会调用subscribeOn中生成的onSubscribe的call方法。对应的就是OperatorSubscribeOn的call，我们来看看这里执行了什么。

@Override
        public void call(final Subscriber<? super T> subscriber) {
            final Worker inner = scheduler.createWorker();
            subscriber.add(inner);
    
            inner.schedule(new Action0() {
                @Override
                public void call() {
                    final Thread t = Thread.currentThread();
    
                    Subscriber<T> s = new Subscriber<T>(subscriber) {
                        @Override
                        public void onNext(T t) {
                            subscriber.onNext(t);
                        }
    
                        @Override
                        public void onError(Throwable e) {
                            try {
                                subscriber.onError(e);
                            } finally {
                                inner.unsubscribe();
                            }
                        }
    
                        @Override
                        public void onCompleted() {
                            try {
                                subscriber.onCompleted();
                            } finally {
                                inner.unsubscribe();
                            }
                        }
    
                        @Override
                        public void setProducer(final Producer p) {
                            subscriber.setProducer(new Producer() {
                                @Override
                                public void request(final long n) {
                                    if (t == Thread.currentThread()) {
                                        p.request(n);
                                    } else {
                                        inner.schedule(new Action0() {
                                            @Override
                                            public void call() {
                                                p.request(n);
                                            }
                                        });
                                    }
                                }
                            });
                        }
                    };
    
                    source.unsafeSubscribe(s);
                }
            });
        }

这里首先生成了一个Worker，看源码得知Worker是一个ScheduledThredPool线程池，不了解线程池的可以先百度一下。然后在线程池中执行接下来的代码，切换到了io线程，然后new Subscriber对象s，最后调用这行代码。

source.unsafeSubscribe(s);

这里的source是第一个create生成的obeservable。最终调用了

hook.onSubscribeStart(this, onSubscribe).call(subscriber);

回到了第一个obeservable的onSubscribe的call方法中，传入的Subscriber对象就是s了，s的onNext方法最终也会执行最终subscribe的onNext方法，至此subscribeOn完毕。