java - RX java 在添加项目时发出延迟的列表项目

Question

我有一个正在从外部源更新的列表。我想要的是一个可观察的，它随着更多项目添加到列表中而以固定延迟发出单个项目。我想用 RXJava 来做到这一点。这是我想出的示例 sudo 代码:

class ItemDispatcher {

  private List<Item> items = new ArrayList();
  private Observable<Item> observable = <some observable>

  private void addItems(List<Item> newItems) {
      items.addAll(newItems);
      // dispatch new items to observable and remove items as they are dispatched 
      //(Items needs to be emitted with 500 milliseconds delay)
  }

  public Observable<Item> getObservable() {
      return observable;
  }
}


//caller
ItemDispatcher dispatcher = new ItemDispatcher();

// when someone calls dispatcher.addItems(...) those items needs to be emitted to subscriber with 500 milliseconds delay

dispatcher.getObservable().subscribe() item -> {
  System.out.println("I'm getting items one by one as they are added to list with 500 millisecond delay")
}

Answer 1

试试这个

class ItemDispatcher {

    private List<Item> items = Collections.synchronizedList(new ArrayList());
    private PublishSubject<List<Item>> itemsSubject = PublishSubject.create();

    private void addItems(List<Item> newItems) {
        items.addAll(newItems);
        itemsSubject.onNext(newItems);
    }

    public Observable<Item> getObservable() {
        return itemsSubject.delay(500, TimeUnit.MILLISECONDS)
                .doOnNext(newItems -> items.removeAll(newItems))
                .flatMapIterable(newItems -> newItems);
    }
}

此外，如果您在其他地方不需要项目，您可以删除 items 字段