rx-java - 如何使用 Observable.from(Future)？

Question

我正在 Android 上学习 RxJava。我想将它与网络一起使用，所以我不能在主线程上做任何事情。

我有以下代码:

final String RXTAG = "Rx";
Log.d(RXTAG, "Starting Rx experiment");

final FutureTask<String> future = new FutureTask<>(new Callable<String>() {
    @Override
    public String call() throws InterruptedException {
        Log.d(RXTAG, "Callable called on thread " + Thread.currentThread().getName());
        Utils.assertNotUIThread();
        Thread.sleep(TimeUnit.SECONDS.toMillis(1)); // Simulates network latency
        return "hello";
    }
});

Observable.from(future, Schedulers.io()).timeout(5, TimeUnit.SECONDS).subscribe(
        new Action1<String>() {
            @Override
            public void call(final String s) {
                Log.d(RXTAG, "Next " + s);
            }
        },
        new Action1<Throwable>() {
            @Override
            public void call(final Throwable throwable) {
                Log.w(RXTAG, throwable);
            }
        },
        new Action0() {
            @Override
            public void call() {
                Log.d(RXTAG, "Completed");
            }
        }
);

但它在 5 秒后以 TimeoutException 结束，Callable called 日志从未显示。出了什么问题，如何让它发挥作用？

Answer 1

尝试在订阅前添加以下代码。

.doOnSubscribe(disposable -> future.run())
.subscribeOn(Schedulers.io())

Rxjava2代码:

Observable.fromFuture(future, Schedulers.io())
.doOnSubscribe(disposable -> future.run())
.subscribeOn(Schedulers.io())
.subscribe(someConsumer()). 

运行良好。