java - 使用 RxJava 在循环中链接异步方法

Question

我有这段代码，我想知道 RxJava 是否可以实现:

• 从服务器查询用户的函数(异步)
• 服务器返回一个用户 JSON 对象，其中包含关联用户配置文件的 ID 列表
• 然后对于每个 ID，它需要获取给定 ID 的 UserProfile(也是异步的)
• 对于每个异步获取的 UserProfile 将其附加到 User 对象，下面是我的伪代码。
• 我不能使用任何阻塞代码，所有请求都应该是异步的。

代码如下:

@Override
public Single<User> retrieve(String entityId) {
    BaasUser baasUser = new BaasUser();
    baasUser.setEntityId(entityId);
    baasUser.setIncludes(Arrays.asList("userProfile"));
    return baasUser.retrieve().map(user -> {
        String username = user.getUsername();
        String dateCreated = user.getDateCreated();
        String dateUpdated = user.getDateUpdated();

        List<UserProfile> userProfiles = new LinkedList<>();

        BaasLink userProfileLink = user.getFirstLink();
        userProfileLink.getEntities().forEach(stubEntity -> {
            Single<UserProfile> observable = stubEntity.retrieve().map(entity -> {
                UserProfile userProfile = new UserProfile();
                userProfile.setEntityId(entity.getStringProperty("entityId"));
                userProfile.setName(entity.getStringProperty("name"));
                return userProfile;
            });
            observable.subscribe(userProfile -> {
                // until all UserProfile is fetched this 'retrieve' "callback" should not return
                userProfiles.add(userProfile);
            }, error -> {
               // err
            });
        });
        User user1 = new User();
        user1.setDateCreated(dateCreated);  
        user1.setDateUpdated(dateUpdated);  
        user1.setUsername(username);
        user1.setUserProfiles(userProfiles);
        return user1;
    });
}

Answer 1

这里有一个示例，说明如何完成您的工作，可能会出现任何类型错误，因为我没有您的对象

    Single.just("user")
            .observeOn(Schedulers.io())
            .flatMap(user -> Observable.zip(getCurrentUserData(user),getUserProfiles(user),(t1, t2) -> {
                //first func will return user eith some data next function wll return List<UserProfile> userProfiles
                return newuser;
            }))
            .subscribeOn(Schedulers.io())
}

private Single<List<UserProfile>> getUserProfiles(User user) {
    Observable.fromIterable( user.getFirstLink().getEntities())
            .flatMap(stubEntity ->stubEntity.retrieve())
            .map(o -> {
                //set user profile data
                return userprofile
            })
            .toList();


}

private Single<User> getCurrentUserData(User user) {
    Observable.just(user)
            .map(s -> {
                //set data
                return user;
            })
}