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

Question

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

下面的代码是@a.bertucci在这里提出的，它使用Android上的RxJava以固定的时间间隔在UI中发出绘制对象，演示了如何以固定的时间间隔发出项目。

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

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

此运算符将采用计算延迟的函数来应用于发射您的下一个项目

 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);
                }
            }
        };
    }
}
}

Answer 2

你可以用

 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来控制速度。