java - 将多个异步 Observable<List> 减少为一个 Observable<List>

Question

所以这就是我想做的:
对于每个事件，我想调用 Service.getList() 10 次，然后将 10 个列表合并为一个。

现在我尝试了这两种方法，它们都在 UT 中工作，但在实际应用程序中失败(我猜我在实际应用程序中给定异步 http 调用时没有正确执行归约操作)。 对于这两种情况，我都看不到 reduce() 之后的日志，onNext() 和 onError() 都看不到，所以我猜测 reduce() 操作未完成。

尝试#1:

public Observable<List<Event>> getEventsForLocation(Location location) {
List<Observable<List<Event>>> obs = new ArrayList<>();
for (Venue v : location.getVenues()) {
     obs.add(getEventsForVenue(v)); //does one http call, returns Observable<List<Event>> 
}
return Observable.concat(Observable.from(obs))
            .reduce((List<Event>) new ArrayList<Event>(), (events, events2) -> {
                events.addAll(events2);
                return events;
            })
            .doOnNext(events -> Log.d("reduce ", events.toString()))
            .doOnError(throwable -> Log.e("reduce error", throwable.toString()));}

尝试#2:

public Observable<List<Event>> getEventsForLocation(Location location) {
return Observable
        .from(location.getVenues())
        .flatMap(venue -> getEventsForVenue(venue)) //does one http call, returns Observable<List<Event>> 
        .reduce((List<Event>) new ArrayList<Event>(), (events, events2) -> {
            events.addAll(events2);
            return events;
        })
       .doOnNext(events -> Log.d("service", "total events " + events.toString()))
       .doOnError(t -> Log.e("service", "total events error2 " + t.toString()));}

以及两种方法的 UT:

    @Test
    public void getEventsForLocation() {
        Location loc = new Location("test", newArrayList(new Venue("v1", "url1"),new Venue("v2", "url2")));

        when(httpGateway.downloadWebPage(Mockito.anyString())).thenReturn(
                Observable.just(readResource("eventsForVenue1.html")),
                Observable.just(readResource("eventsForVenue2.html"))
        );

        TestSubscriber<List<Event>> probe = new TestSubscriber<>();
        service.getEventsForLocation(loc).subscribe(probe);

        probe.assertNoErrors();

        //assert the next event containts contents of all lists
        List<Event> events = probe.getOnNextEvents().get(0);

        //first list
        Assert.assertEquals("Unexpected title", "event1", events.get(0).getName());
        Assert.assertEquals("Unexpected artist", "artist1", events.get(0).getArtist());
        //second list
        Assert.assertEquals("Unexpected title", "event2", events.get(1).getName());
        Assert.assertEquals("Unexpected artist", "artist2", events.get(1).getArtist());
    }

更新

这里是更完整的代码，以及调度程序。

Observable
                .just(loc)
                .subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .flatMap(location -> service.getEventsForLocation(location))
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(getObserver();

Answer 1

如果你不关心列表中的最终顺序，你可以只使用 from + flatMap + flatMapIterable + toList:

Observable.from(location.getVenues())
.flatMap(venue -> getEventsForVenue(venue))
.flatMapIterable(list -> list)
.toList();

如果顺序很重要并且您想“并行”执行 getEventsForVenue，您可以用 concatMapEager 替换 flatMap:

Observable.from(location.getVenues())
.concatMapEager(venue -> getEventsForVenue(venue))
.concatMapIterable(list -> list)
.toList();