java - 使用 RxJava 链接一系列操作 - 下一步去哪里？

Question

场景

我已经构建了 an API对于位于具有请求限制的网关后面的我的应用程序。在我构建 API 之前，我的应用程序自行协调请求，因此可以在几毫秒内启动许多请求，以便在用于获取数据的 9 个提供程序之间同步应用程序的数据。现在，此逻辑已被插入我的 API 适配器层，我需要考虑如何控制每秒请求数以避免达到我自己的速率限制。提高速率限制不是一种选择，因为它需要网关提供商的层级提升，而我不愿意为此付费。

我已经开始使用 RxJava

为了在 Java 社区内的这一强大运动中提高技能，我选择使用 RxJava，以及 Retrofit 和 Retrolamba 来构建我构建的 API SDK。这在很大程度上是成功的，并且可以正常运行。

我的应用

现在，my app允许用户保存“点”，同步检索该地区的本地天气、潮汐和海浪情况。每个点使用 4 个 API 资源来获取完整的数据集，具体而言；

/luna/locations/xtide/{id} - Luna Event detail (read: tide times)
/solar/locations/xtide/{id} - Solar Event detail (read: sunrise/sunset)
/water/locations/{provider}/{id}{?daysData} - Water Event detail (read: swell measures)
/meteo/wwo/weather{?query,daysData} - Meteo Event detail (read: weather data)

该应用程序允许任意数量的广告位，n 这意味着使用当前代码我每个广告位有 4n 个请求。例如，如果我保存了 10 个点并尝试同步所有点 - 我将在大约 0.75 秒内触发 4*10 = 40 个 API 请求!

self 节流

我想使用 Rx 来简化 self 节流我的 API 请求的过程。这是我想要实现的目标的(希望是准确的)大理石图表；

图 1:显示所需流组成的大理石图

SynchronisationService.java 代码看起来有点像这样；

    Observable.zip(
        Observable.from(spots),
        Observable.interval(SYNC_TICK, TimeUnit.MILLISECONDS),
        (obs, timer) -> obs)
        .subscribeOn(scheduler)
        .observeOn(scheduler)
        .unsubscribeOn(scheduler)
        .flatMap(spot -> syncTidePosition.get().buildObservable(spot))
        .subscribe(spotAndTideEvent -> new TideEventSubscriber(
            lunaEventService,
            synchronisationIntentProvider.get(),
            spotAndTideEvent.spot,
            String.format(
                getString(string.tide_error_message),
                spotAndTideEvent.spot.getTidePosition()
            ),
            errorHandlerService,
            localBroadcastManager)
        );

...“buildObservable”调用如下所示；

Observable<SpotAndTideEventTuple> buildObservable(final Spot spot) {
    return Observable.zip(
        Observable.just(spot),
        lunaEventsProvider
            .listTideTimes(
                spot.getTideOperator(),
                Integer.toString(spot.getTidePosition())
            ),
        SpotAndTideEventTuple::new
    );
  }

...lunaEventsProvider.listTideTimes(...) 方法看起来像；

public Observable<List<TideEvent>> listTideTimes(@NonNull final LunaProvider provider,
                                                   @NonNull final String identifier) {
    return getRetrofitServiceImpl(LunaEventsProviderDefinition.class)
        .listTideTimes(provider, identifier)
        .map(TideEventsTemplate::buildModels);
  }

问题

作为 Rx 业余爱好者，我已经阅读了很多文档以达到此目的，但是在遇到代码错误时，我不知道下一步该怎么做。订阅不会导致发射开始(如所示的 fragment )，或者如果我稍微调整一下，我会得到一个无用的低级 NPE (rx.Scheduler)。

我应该从这里去哪里？对于所描述的场景，我是否在正确的轨道上使用 Rx？任何帮助表示赞赏。

Answer 1

有点尴尬的是，我看到的 NPE 错误与 Rx 无关，而是我指定用于运行操作的调度程序被注入(inject)到 android.app.Service 中，但由于轻微的“配置错误”(省略了 @Inject 注释!) scheduler 变量为空。

知道我错过这个的原因是因为我的 Scheduler 注入(inject)也是合格的，这意味着它“看起来”与我在类；

@Inject @IoScheduler Scheduler scheduler;
@Inject LocalBroadcastManager localBroadcastManager;
@Inject NotificationManager notificationManager;
@Inject SharedPreferences sharedPrefs;

好吧，我在构建那些弹珠图和分解我对 Rx 的理解时玩得很开心。当前调用现在协调所有 4 个 API 请求，如下所示；

    Observable.zip(
        Observable.from(spots),
        Observable.interval(SYNC_TICK, TimeUnit.MILLISECONDS),
        (obs, timer) -> obs)
        .subscribeOn(scheduler)
        .observeOn(scheduler)
        .unsubscribeOn(scheduler)
        .flatMap(this::buildObservable)
        .subscribe(
            new EventSubscriber(
                lunaEventService,
                solarService,
                swellService,
                conditionsService,
                synchronisationIntentProvider.get(),
                errorHandlerService,
                localBroadcastManager,
                TRENDING_LENGTH_DAYS
            )
        );

这是此服务重构的一部分，因此我希望它能有所改变，尤其是在将测试置于绿色之下时。很高兴我坚持了下来，每次我学习一个函数时，使用 Rx 确实删除了大约 50 到 100 行代码!