java - RxJava 将新对象添加到流中

Question

我正在尝试将新对象注入可能存在的流中。

假设我有这个：

ItemUtils.java

public static void processItems(List<Item> items) {
    Observable.from(items)
        .subscribeOn(Schedulers.io())
        .flatMap(ItemUtils::doSomeHeavyProcessing)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe();
}

在另一个班级：

List<Item> items = new ArrayList<>();

public void onClick() {
    processItems(list);
}

---

每次不耐烦的用户按下按钮时，是否可以将项目注入一个独特的项目流？

Answer 1

有可能，这个问题缺乏精确性，所以这个答案很笼统。

基本思想是使用 [ concatMap]( http://reactivex.io/RxJava/javadoc/rx/Observable.html#concat(rx.Observable , rx.Observable)) 或此运算符的变体。

因此，例如，取决于代码真正需要做什么，它可能是：

public static void processItems(List<Item> items) {
    Observable<ProcessdItem> processedItems = Observable.from(items)
        .subscribeOn(Schedulers.io())
        .flatMap(ItemUtils::doSomeHeavyProcessing)
        .observeOn(AndroidSchedulers.mainThread());

    Observable.concat(
        Observable.just(ProcessedItems.preProcessed()),
        processedItems
    ).subscribe();
}

或者在嵌套的 observable 中。

public static void processItems(List<Item> items) {
    Observable.from(items)
        .subscribeOn(Schedulers.io())
        .flatMap(i ->
            Observable.just(ProcessedItems.preProcessed())
                .concatWith(ItemUtils::doSomeHeavyProcessing))
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe();
}

Answer 2

如果 List items 是一种队列，用户可以在 processItems 已经启动时添加更多 Items，您应该使用Subjects

Queue<Item> items = new LinkedList<>();
PublishSubject<Item> subject = PublishSubject.create(); 

...
//subscribe once and supply new items in onClick
subject.subscribeOn(Schedulers.io())
    .flatMap(ItemUtils::doSomeHeavyProcessing)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe();

...
public void onClick() {
    while(!queue.isEmpty()){
        subject.onNext(queue.remove());
    }
}