如何使用Observable.from(将来)？ [英] How to use Observable.from(Future)?

问题描述

我正在学习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结束，并且不会显示名为的可调用日志。出了什么问题，以及如何使其正常工作？

推荐答案

您必须自己运行Future。但不幸的是，这会导致在主线程上运行Callable，否则您必须使用Executor。

相反，我提供了以下解决方案：

// Extract Callable from FutureTask

Single.fromCallable(callable).timeout(5, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).subscribe(
    new Action1<String>() {
        @Override
        public void call(final String s) {
            Log.d(RXTAG, "Success " + s);
        }
    },
    new Action1<Throwable>() {
        @Override
        public void call(final Throwable throwable) {
            Log.w(RXTAG, throwable);
        }
    }
);

现在运行正常：

07-29 13:41:26.516 D/Rx: Starting Rx experiment
07-29 13:41:26.547 D/Rx: Callable called on thread RxIoScheduler-2
07-29 13:41:27.550 D/Rx: Success hello

这篇关于如何使用Observable.from(将来)？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！