在 RxJava 中以取决于发射项目本身的间隔发射项目

Question

在 RxJava for Android 中，我想在取决于项目本身的间隔内发出项目：在 Observable 中，我从队列中拉出一个项目，处理它并发出它。根据项目的类型，我想调整下一个项目的发射时间（减慢或加快间隔）。

@a.bertucci 在此处Emit objects for drawing in the UI in a regular interval using RxJava on Android 提出的以下代码演示了如何定期发射项目。

private void drawPath(final String chars) {
    Observable.zip(
        Observable.create(new Observable.OnSubscribe<Path>() {
            // all the drawing stuff here
            ...
        }),
        Observable.timer(0, 50, TimeUnit.MILLISECONDS),
        new Func2<Path, Long, Path>() {
            @Override
            public Path call(Path path, Long aLong) {
                return path;
            }
        }
    )
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    ...
}

我现在的问题是，是否有可能在可观察对象发射时修改发射频率，以及使用 RxJava 的首选实现是什么。

Answer 1

你可以使用

 public final <U, V> Observable<T> delay(
            Func0<? extends Observable<U>> subscriptionDelay,
            Func1<? super T, ? extends Observable<V>> itemDelay)

或

public final <U> Observable<T> delay(Func1<? super T, ? extends Observable<U>> itemDelay)

你可以使用itemDelay来控制速度。

Answer 2

根据您的评论，我认为您应该为此建立一个新的运营商。

该运算符将采用一个函数来计算应用到发出下一个项目的延迟

 Observable.range(1, 1000).lift(new ConfigurableDelay((item) -> 3 SECONDS)
                          .subscribe();

你可以试试这样的：

public class ConfDelay {

public static void main(String[] args) {
    Observable.range(1, 1000).lift(new ConfigurableDelay(ConfDelay::delayPerItem, Schedulers.immediate()))
            .map((i) -> "|")
            .subscribe(System.out::print);
}


public static TimeConf delayPerItem(Object item) {
    long value = ((Integer) item).longValue();
    return new TimeConf(value * value, TimeUnit.MILLISECONDS);
}

private static class TimeConf {
    private final long time;
    private final TimeUnit unit;

    private TimeConf(final long time, final TimeUnit unit) {
        this.time = time;
        this.unit = unit;
    }
}

private static class ConfigurableDelay<T> implements Observable.Operator<T, T> {
    private final Func1<T, TimeConf> itemToTime;
    private final Scheduler scheduler;

    public ConfigurableDelay(final Func1<T, TimeConf> itemToTime) {
        this(itemToTime, Schedulers.computation());
    }

    public ConfigurableDelay(final Func1<T, TimeConf> itemToTime, final Scheduler scheulder) {
        this.itemToTime = itemToTime;
        this.scheduler = scheulder;
    }

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
        return new Subscriber<T>(subscriber) {

            private TimeConf nextTime = null;

            @Override
            public void onCompleted() {
                subscriber.onCompleted();
            }

            @Override
            public void onError(final Throwable e) {
                subscriber.onError(e);
            }

            @Override
            public void onNext(final T t) {
                TimeConf previousNextTime = nextTime;
                this.nextTime = itemToTime.call(t);
                if (previousNextTime == null) {
                    subscriber.onNext(t);
                } else {
                    scheduler.createWorker().schedule(() -> subscriber.onNext(t), previousNextTime.time, previousNextTime.unit);
                }
            }
        };
    }
}
}