java - RxJava 2 - 在 concatMap 中减少队列

Question

我有一个发出一些数据的主题。然后我想修改队列中的这个数据（长操作）。但是当另一个主题会发出其他东西时，我想减少我的队列

    PublishSubject<Integer> myBaseSubject = PublishSubject.create();
    PublishSubject<Integer> otherSubject = PublishSubject.create();

    myBaseSubject
            .concatMap(integer -> Observable.timer(5, TimeUnit.SECONDS) // long operation
                    .map(aLong -> integer)
            )
            .subscribe(
                    integer -> Log.i(TAG, "result: " + integer),
                    Functions.emptyConsumer()
            );

    myBaseSubject.onNext(1);
    myBaseSubject.onNext(2);
    myBaseSubject.onNext(3);
    myBaseSubject.onNext(4); 
    myBaseSubject.onNext(5);

    otherSubject.onNext(3);
    otherSubject.onNext(4);

    myBaseSubject.onNext(6);
    myBaseSubject.onNext(7);

otherSubject现在，如果这些元素尚未被concatMap（它们仍在队列中）消耗，我想修改我的一次性以跳过由（3 和 4）发出的元素。

有可能实现吗？

PS: PublishSubject 不是必需的 - 它只是为了简单起见

Answer 1

您可以ConcurrentHashMap跟踪哪些元素不应由concatMap. 我假设在创建该项目的索引之后决定不处理该项目。

ConcurrentHashMap<Integer, Integer> map = new ConcurrentHashMap<>();

myBaseSubject
    .doOnNext(v  -> map.put(v, v))
    .concatMap(integer -> {
        if (map.remove(integer) != null) {
            return Observable.timer(5, TimeUnit.SECONDS) // long operation
                .map(aLong -> integer)
        }
        return Observable.empty();
    })
    .subscribe(
        integer -> Log.i(TAG, "result: " + integer),
        Functions.emptyConsumer()
    );

myBaseSubject.onNext(1);
myBaseSubject.onNext(2);
myBaseSubject.onNext(3);
myBaseSubject.onNext(4); 
myBaseSubject.onNext(5);

map.remove(3);
map.remove(4);

myBaseSubject.onNext(6);
myBaseSubject.onNext(7);