android - RxJava 链式 Observable 和 NetworkMainThreadException

Question

所以我有这段代码:

public Observable<AbstractXMPPConnection> connect(final AbstractXMPPConnection connection) {
    return Observable.<AbstractXMPPConnection>create(subscriber -> {
        try {
            AbstractXMPPConnection connection2 = connection.connect();
            if (connection2.isConnected()) {
                subscriber.onNext(connection2);
                subscriber.onCompleted();
            }
        } catch (SmackException | IOException | XMPPException e) {
            e.printStackTrace();
            subscriber.onError(e);
        }
    })
    .doOnError(throwable -> LOGI("111", "Connection OnError called"));
}


public Observable<AbstractXMPPConnection> connectWithRetry(final AbstractXMPPConnection connection) {
       return connect(connection)
               .retryWhen(attempts -> attempts.zipWith(Observable.range(1, MAX_CONNECTION_TRIES), (throwable, integer) -> new Pair<>(throwable, integer))
                       .flatMap(pair -> {
                           if (pair.second == MAX_LOGIN_TRIES)
                               return Observable.error(pair.first);
                           return Observable.timer(pair.second, TimeUnit.SECONDS);
                       }));
    }


public void connect() {
        assertTrue("To start a connection to the server, you must first call init() method!",
                this.connectionConfig != null);

        connectionHelper.connectWithRetry(connection)
                .observeOn(Schedulers.newThread())
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<AbstractXMPPConnection>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                        LOGI(TAG, "ConnectionHelper Connection onError\n");

                        /**{@link LoginActivity#onConnectionFailure(OnConnectionFailureEvent)} */
                        MainApplication.getInstance().getBusInstance().post(new OnConnectionFailureEvent());
                    }

                    @Override
                    public void onNext(AbstractXMPPConnection connection) {
                        LOGI(TAG, "ConnectionHelper Connection onNext");
//                        onConnected();
                    }
                });
    }

我对链接可观察对象有一些疑问。想象一下这个场景，我有一个连接 Observable，有时我会使用它，但我主要使用 connectWithRetry() Observable。

我的问题是，如果添加这个会发生什么:

.observeOn(Schedulers.newThread())
.subscribeOn(AndroidSchedulers.mainThread())

对于 connect() 和 connectWithRetry()？在这种情况下，当我打电话public void connect 并指定一个scheduler，前面的被忽略了？

为什么我得到 NetworkOnMainThreadException？明确的 observeOn(Schedulers.newThread()) 在那里，它不应该给我那个错误

Answer 1

我会先解决您的 NetworkOnMainThread 问题。

observeOn(Schedulers.newThread()) 表示输出将在新线程上被观察 - 即您订阅者中的代码 (onComplete/Error/Next) 将在该线程上运行。

subscribeOn(AndroidSchedulers.mainThread() 意味着订阅 将发生在主线程上 - 您创建的可观察对象中的代码 (connection.connect() 等)是订阅发生时运行的代码。

所以简单地交换调度器:

.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())

因此，为了解决您的第一个问题，它们并没有被忽略，它们只是使用不当。希望从这里您可以看到，如果您将类似的调用移动到返回可观察对象的方法中的链中，将会发生什么:与您已经完成的没有什么不同。这些电话只是在不同的地方。

那么调度程序选择放在哪里呢？随你(由你决定。您可以通过不在创建可观察对象的方法中调用subscribeOn 来提高清晰度:

 connectionHelper.connectWithRetry(connection)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())

但是，如果您觉得无缘无故到处调用它，则可以将 subscribeOn 调用移到您的方法中:

return connect(connection)
           .retryWhen(...)
           .flatMap(...)
           .subscribeOn(Schedulers.io())
           .observeOn(AndroidSchedulers.mainThread());

请注意，这些不必像这样 bundle 在一起 - 您可以在方法中使用 subscribeOn，但将 observeOn 留给任何需要其结果的调用者在特定线程上。