RxJava 将一个 Observable 拆分为两个 subObservable

Question

我对RxJava完全陌生，我花了一整天的时间来理解它，我正在思考如何解决这个问题：

我有一个对象，由 Retrofit 获取，它包含两个 ArrayLists，我必须以不同方式处理每个 ArrayList。目前它看起来像：

apiService.getUser(token).enqueue(new Callback<User>() {
            @Override
            public void onResponse(Response<User> response) {

                final User user = response.body();

                for (Skill s : user.getSkills()) {
                    // process here first ArrayList
                }

                for (OrganizerAction o : user.getOrganizerActions()) {
                    // process here second ArrayList
                }
            }

            @Override
            public void onFailure(Throwable t) {
                t.printStackTrace();
            }
        });

更新：

public class User {

    // fields

    @SerializedName("organizer_actions")
    @Expose
    private List<OrganizerAction> mOrganizerActions;

    @SerializedName("skills")
    @Expose
    private List<Skill> mSkills;

    public List<OrganizerAction> getOrganizerActions() {
        return mOrganizerActions;
    }

    public List<Skill> getSkills() {
        return mSkills;
    }
}

谢谢，
安东

Answer 1

此答案适用于 Retrofit 2.0.0-beta，您似乎正在使用它。此外，您没有提供 POJO 或服务定义，因此将使用通用 GitHub API 示例作为指导，修改以匹配您的指定数据。

第一步是将您的服务定义转换为使用 Observable 而不是 Call。

public interface GitHubService {
    @GET("/users/{user}")
    Observable<User> getUser(@Path("user") String user);
}

其中User是

public class User {
    public String login;
    public int id;
}

接下来，使用 addCallAdapterFactory --

将自定义调用适配器添加到改造构建器

Retrofit retrofit = new Retrofit.Builder()
            .baseUrl("https://api.github.com/")
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .build();

以常规方式获取服务 --

GitHubService gitHubService = retrofit.create(GitHubService.class);

接下来，获取您的可观察对象并对其调用 cache 以创建一个 Observable 来重放结果。从那个 Observable 中，您可以订阅多次，在您的情况下，您可以订阅两次。对于您感兴趣的每种数据类型一次，并使用 map function to transform from the User object to your specific fields. map allows you to apply function to the data in the observable. See the docs 了解更多详细信息。在这个例子中，我们将制作两个流。 id 和 login 字段各一个。

Observable<User> getUserResult = gitHubService.getUser("octocat").cache(1);

getUserResult.map(new Func1<User, Integer>() {
    @Override
    public Integer call(User user) {
        return user.id;
    }
}).subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer id) {
        Log.d("Stream 1", "id = " + id);
    }
});

getUserResult.map(new Func1<User, String>() {
    @Override
    public String call(User user) {
       return user.login;
    }
}).subscribe(new Action1<String>() {
    @Override
    public void call(String login) {
        Log.d("Stream 2", "login = " + login);
    }
});

最后，确保您的 gradle 文件具有所需的依赖项，

compile 'io.reactivex:rxjava:1.0.14'
compile 'com.squareup.retrofit:retrofit:2.0.0-beta1'
compile 'com.squareup.retrofit:adapter-rxjava:2.0.0-beta1'
compile 'com.squareup.retrofit:converter-gson:2.0.0-beta1'

并且，与您的问题没有直接关系，但是如果您打算在 Android 中使用 RxJava，我建议您查看 Retrolambda（如果您还没有）。上面的映射和订阅代码，以及一般的 Rx 代码，使用 lambda 更简洁。

getUserResult.map(user -> user.id).subscribe(
        id -> { Log.d("Stream 1", "id = " + id); }
);

getUserResult.map(user -> user.login).subscribe(
        login -> { Log.d("Stream 2", "login = " + login); }
);

RxJava 将一个 Observable 拆分为两个 subObservable

RxJava split one Observable to two subObservables

java

android

rx-java

retrofit

rx-android