android - 延迟除了第一次 Rxjava Android

Question

我正在进行异步调用，10 秒后 1 分钟，这意味着将进行大约 6 次调用，但问题在于我希望它在特定的 条件 上应用 delay

Observable
.just(listOfSomethings_Locally)
.take(1, TimeUnit.MINUTES)
.serialize()
.delaySubscription( // this is confusing part 
() ->
    Observable.just(listOfItems_Network).take(10,TimeUnit.SECONDS)
) 

我想要的是除第一次调用外延迟网络调用 10 秒，并在 10 秒后取消网络调用，所以我应该在 1 分钟内有 6 个调用。

编辑

由于场景困惑这里重新定义了场景:

what i have is large list of drivers locally and i want to send request to each of them after every 10 seconds and listen to another subscriber to check if driver didn't canceled it within 10 seconds , this process will go for about 1 minute, if one driver cancel i should immediately send request to next one

到目前为止编写的代码:

Observable.from(driversGot)
                .take(1,TimeUnit.MINUTES)
                .serialize()
                .map(this::requestRydeObservable) // requesting for single driver from driversGot (it's a network call)
                .flatMap(dif ->
                        Observable.amb(
                                kh.getFCM().driverCanceledRyde(), // listen for if driver cancel request returns integer
                                kh.getFCM().userRydeAccepted()) // listen for driver accept returns RydeAccepted object
                                .map(o -> {
                                    if (o instanceof Integer) {
                                        return new RydeAccepted();
                                    } else if (o instanceof RydeAccepted) {
                                        return (RydeAccepted) o;
                                    }
                                    return null;
                                }).delaySubscription(10,TimeUnit.SECONDS)
                )
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(fua -> {
                    if (fua == null) {
                        UiHelpers.showToast(context, "Invalid Firebase response");
                    } else if (!fua.getStatus()) { // ryde is canceled because object is empty
                        UiHelpers.showToast(context, "User canceled ryde");
                    } else { // ryde is accepted
                        UiHelpers.showToast(context, "User accepted ryde");
                    }
                }, t -> {
                    t.printStackTrace();
                    UiHelpers.showToast(context,"Error sending driver requests");
                }, UiHelpers::stopLoading);

Answer 1

对您的代码的反馈

你不需要 take 和 serialize 因为 just 会立即发出东西 + 已经串行了。

delaySubscription 似乎是一个奇怪的选择，因为在传递的 observable 生成一个事件之后，进一步的事件不会被延迟(这与你的情况相矛盾)

选项 #1，仅 rx

使用 delay + 计算其余事件的个别延迟(因此第一个延迟 0 秒，第二个延迟 1 秒，第三个延迟 3，...)

            AtomicLong counter = new AtomicLong(0);
    System.out.println(new Date());
    Observable.just("1", "2", "3", "4", "5", "6")
        .delay(item -> Observable.just(item).delay(counter.getAndIncrement(), TimeUnit.SECONDS))
        .subscribe(new Consumer<String>() {
            public void accept(String result) throws Exception {
                System.out.println(result + " " + new Date());
            }
        });        
        System.in.read();

选项 #2:速率限制

您的用例似乎适合速率限制，因此我们可以使用 Guava 中的 RateLimiter:

            RateLimiter limiter = RateLimiter.create(1);
    System.out.println(new Date());
    Observable.just("1", "2", "3", "4", "5", "6")
        .map(r -> {
            limiter.acquire();
            return r;
        })
        .subscribe(new Consumer<String>() {
            public void accept(String result) throws Exception {
                System.out.println(result + " " + new Date());
            }
        });        
        System.in.read();

两者的工作方式相似:

Tue Nov 15 11:14:34 EET 2016
1 Tue Nov 15 11:14:34 EET 2016
2 Tue Nov 15 11:14:35 EET 2016
3 Tue Nov 15 11:14:36 EET 2016
4 Tue Nov 15 11:14:37 EET 2016
5 Tue Nov 15 11:14:38 EET 2016
6 Tue Nov 15 11:14:39 EET 2016

如果您有要求，速率限制器会更好地工作，例如处理 5 秒，然后它将允许下一个请求更快地补偿延迟并达到 10 秒 1req/s 的目标。