如何在 RxJava 中延迟从列表中发出项目？答案

【问题标题】：How to emit items from a list with delay in RxJava?如何在 RxJava 中延迟从列表中发出项目？
【发布时间】：2015-03-26 13:58:43
【问题描述】：

我正在使用 Retrofit 从 REST API 获取书签：

public interface BookmarkService {
    @GET("/bookmarks")
    Observable<List<Bookmark>> bookmarks();
}

现在我想延迟发出这个列表中的每个项目。

我在 Java 中做了类似于 this 的操作，但从未触发过 onCompleted。

private Observable<Bookmark> getBookmarks() {
    return getBookmarkService().bookmarks()
         .flatMap(new Func1<List<Bookmark>, Observable<Bookmark>>() {
            @Override
            public Observable<Bookmark> call(List<Bookmark> bookmarks) {
                Observable<Bookmark> resultObservable = Observable.never();
                for (int i = 0; i < bookmarks.size(); i++) {
                    List<Bookmark> chunk = bookmarks.subList(i, (i + 1));
                    resultObservable = resultObservable.mergeWith(Observable.from(chunk).delay(1000 * i, TimeUnit.MILLISECONDS));
                }

                return resultObservable;
            }
        })
        .observeOn(AndroidSchedulers.mainThread());
}

我做错了什么？

用法：

mSwipeRefreshLayout.setRefreshing(true);
getBookmarks()
    .subscribe(new Observer<Bookmark>() {
        @Override
        public void onCompleted() {
            Timber.i("Completed");
            mSwipeRefreshLayout.setRefreshing(false);
        }

        @Override
        public void onError(Throwable e) {
            Timber.i("Error: %s", e.toString());
            mSwipeRefreshLayout.setRefreshing(false);
        }

        @Override
        public void onNext(Bookmark bookmark) {
            Timber.i("Bookmark: %s", bookmark.toString());
            mBookmarksAdapter.addItem(bookmark);
        }
    });

【问题讨论】：

标签： android retrofit rx-java rx-android

【解决方案1】：

当您使用合并操作时，如果所有 Observable 都已完成，则将调用 onCompleted。但Observable.never() 永远不会完成。请改用Observable.empty()。

根据您的代码，您希望延迟发出子列表。子列表只包含一个元素

你可以做什么：平面化你的列表，发出每个项目。缓冲它以根据项目构建列表，然后使用延迟。

private Observable<Bookmark> getBookmarks() {
    return getBookmarkService().bookmarks()
                               .flatMap((bookmarks) -> Observable.from(bookmarks)
      .buffer(1)
      .scan(new Pair(0, null), (ac, value) -> new Pair(acu.index + 1, value)
      .flatMap(pair -> Observable.just(pair.value).delay(pair.index, SECONDS)) 
                               .observeOn(AndroidSchedulers.mainThread());
}

它可能有效（未经测试）

【讨论】：