避免在发送回调之前使用 CountDownLatch 等待多个线程答案

【问题标题】：Avoid using CountDownLatch to wait for many threads before sending the callback避免在发送回调之前使用 CountDownLatch 等待多个线程
【发布时间】：2016-01-23 02:39:52
【问题描述】：

所以我有一个轨道 ID 列表，对于每个轨道 ID，我需要执行网络请求以获取轨道详细信息，我使用 for 循环来启动所有请求，并使用锁存器等待所有请求完成。完成后，回调将与已填充的曲目列表一起发送。

我想知道是否有更好的方法来做到这一点，也许使用 RxJava ？

我在 Android 中使用 Retrofit 2.0。

    public IBaseRequest batchTracksById(final TrackIdList trackIdListPayload, final IRequestListener<TracksList> listener) {
    final TracksList tracks = new TracksList();
    final Track[] trackArray = newrack[trackIdListPayload.getTrackIds().length];
    tracks.setTrack(trackArray);

    final CountDownLatch latch = new CountDownLatch(trackArray.length);


    Thread t = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                latch.await();
                handler.post(new Runnable() {
                    @Override
                    public void run() {
                        listener.onRequestUpdate(null, tracks, null, true);
                    }
                });

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
    t.start();

    for (String id : trackIdListPayload.getTrackIds()) {
        getTrackById(id, new IRequestListener<Track>() {
            @Override
            public void onRequestFailure(IBaseRequest request, Exception exception) {
                latch.countDown();
            }

            @Override
            public void onRequestUpdate(IBaseRequest request, Track track, RequestState state, boolean requestComplete) {
                //iterate through the tracks and update the ones in the thing

                int i = 0;
                for (String s : trackIdListPayload.getTrackIds()) {
                    if (s.equals(track.getTrackId())) {
                        trackArray[i] = track;
                        // don't break here, as we may have a case where we have multiple instances of the same trackId (although
                        // at the moment a request will be made for each anyway...
                    }
                    i++;
                }

                latch.countDown();
            }
        });
    }

    return null;
}

【问题讨论】：

标签： java android multithreading retrofit rx-java

【解决方案1】：

如果您想异步发出所有请求并等待它们返回，您可以这样做（为了简洁和可读性，使用 lambdas）：

tracks.flatMap(track -> getTrackDetails(track.id)
                          .subscribeOn(Schedulers.io()))
      .toList()
      .doOnNext(list -> processTrackList())
      ...

如果您要求按tracks 的顺序返回结果，但仍被异步请求，那么在即将发布的 rxjava 1.0.15 中，您将能够做到这一点

tracks.concatMapEager(track -> getTrackDetails(track.id)
                          .subscribeOn(Schedulers.io())
      .toList()
      .doOnNext(list -> processTrackList())
      ...

【讨论】：

【解决方案2】：

如果我理解正确，您有一个曲目列表作为输入，并且您想要一个 Web 服务结果列表。如果您可以使您的网络调用同步（rxjava 将为您处理后台处理），这是使用 RxJava 执行此操作的一种简单方法。

 Observable.from(trackList)
                    .map(new Func1<Track, Response>() {
                        @Override
                        public Response call(Track track) {
                            return makeRequestSynchronously(track.id());
                        }
                    })
                    .toList()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Subscriber<List<Response>>() {

                        @Override
                        public void onCompleted() {

                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onNext(List<Response> responses) {

                        }
                    });

编辑：您可以更改 Retrofit 以从 web 服务返回 observables，如果这样做，您需要将地图更改为以下内容

 .flatMap(new Func1<Track, Observable<Response>>() {
                    @Override
                    public Observable<Response> call(Track track) {
                        return makeRequestThatReturnsObservable(track.id());
                    }
                })

【讨论】：