使用RxJava链接一系列操作——下一步该怎么做？

Question

我为我的应用程序构建了一个应用程序接口，它位于一个带有请求限制的网关后面。在我构建应用程序接口之前，我的应用程序会自行协调请求，因此可以在毫秒内启动许多请求，以便在用于获取数据的9个提供程序中同步应用程序的数据。现在，这个逻辑已经被推到我的应用程序接口适配器层。我需要考虑如何控制每秒请求的数量，以避免达到我自己的速率限制。提高速率限制不是一个选项，因为这需要网关提供程序的层冲击，我不愿意为此付费。

为了提高Java社区中这一强大运动的水平，我选择使用RxJava，以及Retrofit和Retrolamba作为我构建的应用编程接口SDK。这在很大程度上是成功的，并且运行正常。

现在，我的应用程序允许用户保存“点”，当同步检索该地区的当地天气、潮汐和涌浪条件时。每个spot使用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个点并试图同步所有-我将导致4*10=40个应用编程接口请求在0.75秒左右被触发！

我想使用Rx来简化自我限制我的应用编程接口请求的过程。这里有一个大理石图表（希望准确）说明我想要实现的目标；

同步服务。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业余爱好者，我已经阅读了很多文档来达到这一目的，但在遇到代码错误时，我不知道下一步该怎么做。要么订阅没有导致排放开始（如所示片段），要么如果我稍微调整一下，就会得到一个无用的低级NPE（rx.Scheduler）。

从这里我应该去哪里？我是否在正确的轨道上使用Rx描述的场景？感谢任何帮助。




Answer 1

匿名用户

令人尴尬的是，我看到的NPE错误与Rx无关，而我指定运行该操作的调度程序正被注入到android中。应用程序。服务但由于轻微的“配置错误”（省略了@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行代码！