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

Question

我有一个从外部源更新的列表。我想要的是一个observable，随着更多项目添加到列表中，它会以固定的延迟发出单个项目。我想用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字段