1

我有一段代码，他的工作是更新本地缓存。此缓存更新有两个触发器：

  1. 以固定间隔
  2. 要求时

所以这是一个关于我如何做到这一点的基本示例。

forceReloadEvents = new SerializedSubject<Long, Long>(PublishSubject.<Long> create());
dataUpdates = Observable
    .merge(forceReloadEvents, Observable.timer(0, pullInterval, TimeUnit.SECONDS))
    .flatMap(new Func1<Long, Observable<Boolean>>() {
        @Override
        public Observable<Boolean> call(Long t) {
            return reloadData(); // operation that may take long
        }
    })
    .publish();

dataUpdates.subscribe();
dataUpdates.connect();

然后我有

public void forceReload() {
    final CountDownLatch cdl = new CountDownLatch(1);

    dataUpdates
        .take(1)
        .subscribe(
            new Action1<Boolean>() {
                @Override
                public void call(Boolean b) {
                    cdl.countDown();
                }
            }
        );

    forceReloadEvents.onNext(-1L);

    try {
        cdl.await();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

这可行，但问题是当我开始有多个并发调用时forceReload()：将不会并发执行reloadData()但元素将排队并且进程将循环重新加载数据，直到发送到的所有事件forceReloadEvents都已被消耗，即使forceReload()已经完成由于之前的事件发布了CountDownLatch.

我想使用onBackPressureDrop，但似乎没有诱发背压，也没有任何东西被丢弃。我想要的是某种强制背压的方法，以便合并了解一次只能处理一个元素，并且必须删除任何后续事件，直到当前执行完成。

我考虑过使用bufferor throttleFirst，但我不想强制每个事件之间有一个特定的时间，我宁愿根据重新加载缓存所需的时间进行自动缩放。您可以将其视为throttleFirst直到reloadData完成。

4

2 回答 2

2

编辑：根据评论，您可以在 flatMap 中使用 AtomicBoolean 作为门，以便在门再次打开之前不开始重新加载：

public class AvoidReloadStorm {
    static Observable<Boolean> reload() {
        return Observable.just(true)
        .doOnNext(v -> System.out.println("Reload started..."))
        .delay(10, TimeUnit.SECONDS)
        .doOnNext(v -> System.out.println("Reloaded"));
    }
    public static void main(String[] args) throws Exception {
        Subject<Long, Long> manual = PublishSubject.<Long>create().toSerialized();
        Observable<Long> timer = Observable.timer(0, 5, TimeUnit.SECONDS)
                .doOnNext(v -> System.out.println("Timer reload"));

        AtomicBoolean running = new AtomicBoolean();

        ConnectableObservable<Boolean> src = Observable
        .merge(manual.onBackpressureDrop(), timer.onBackpressureDrop())
        .observeOn(Schedulers.io())
        .flatMap(v -> {
            if (running.compareAndSet(false, true)) {
                return reload().doOnCompleted(() -> {
                    running.set(false);
                });
            }
            System.out.println("Reload rejected");
            return Observable.empty();
        }).publish();

        src.subscribe(System.out::println);

        src.connect();

        Thread.sleep(100000);
    }
}
于 2015-06-15T15:45:44.950 回答
2

感谢 akarnokd，我完成了这项工作！

这是我根据他的回答创建的解决方案：

Observable<Long> forceReloadEvents = this.forceReloadEvents
    .asObservable()
    .onBackpressureDrop();

Observable<Long> periodicReload = Observable
    .timer(0, pullInterval, TimeUnit.SECONDS)
    .onBackpressureDrop();

final AtomicBoolean running = new AtomicBoolean();

dataUpdates = Observable
    .merge(forceReloadEvents, periodicReload)
    .filter(new Func1<Long, Boolean>() {
        @Override
        public Boolean call(Long t) {
            return running.compareAndSet(false, true);
        }
    })
    .observeOn(Schedulers.io())
    .flatMap(new Func1<Long, Observable<Boolean>>() {
        @Override
        public Observable<Boolean> call(Long t) {
            return reloadData();
        }
    })
    .doOnNext(new Action1<Boolean>() {
        @Override
        public void call(Boolean t) {
            running.set(false);
        }
    })
    .publish();

dataUpdates.subscribe();
dataUpdates.connect();

我不确定onBackpressureDrop这里是否有用，但我将其设置为预防措施。

forceReload 代码不会改变。

于 2015-06-16T08:45:06.403 回答