java - RxJava 2 : Observable. create() 等待异步代码使用 Thread.sleep() 完成

Question

我正在使用 Observable.create(...) 包装一些异步代码。异步代码需要“Activity ”五秒钟，然后应该停止。这就是我目前阻止它的方法:

Observable.<MyObject>create(emitter -> {

    // Some async code

    emitter.setDisposable(Disposables.fromRunnable(() -> {
        // Stop the above async code
    }));

    // Wait for 5 seconds until the async code above has had enough time to finish its task.
    Thread.sleep(5000);

    if (!emitter.isDisposed()) {
        // Stop the above async code
        emitter.setDisposable(null);
    }
})...

我不确定这是否是最好的方法，因为使用了Thread.sleep(5000)。有没有更好的方法来使用一些 RxJava 代码来做到这一点？

Answer 1

take 存在定时重载，允许您在指定时间过去之前从源获取项目:

Observable.<MyObject>create(emitter -> {

    // Some async code

    emitter.setDisposable(Disposables.fromRunnable(() -> {
        // Stop the above async code
    }));
})
.take(5, TimeUnit.SECONDS);

您仍然需要在 create 中设置 Disposable。