如何使用 rxjava 处理一系列异步事件？答案

【问题标题】：How to process a serial of async event by using rxjava?如何使用 rxjava 处理一系列异步事件？
【发布时间】：2016-12-10 05:52:57
【问题描述】：

我是 Rxjava 的新手，我在 google 上搜索了很多。但对我的问题不是很有用。我的情况是使用第三方 SDK 的 Living App。但是进入客厅的过程很复杂，无法控制。

大概的整体流程是这样的：

从应用服务器获取肝脏的房间信息 --- 异步
调用SDK的api ---Async进入客厅
如果成功，调用SDK的IM api---Async加入聊天群
然后使用最后一步的结果更新 UI。

每个步骤都依赖于上一步的结果，并且回调不是相同的 IN/OUT 类型，所以我很难用 rxjava 编写代码。

谢谢。

【问题讨论】：

标签： asynchronous rx-java

【解决方案1】：

您可以使用 Observable.fromEmitter 将异步调用包装在一个可观察对象中，如下所示：

return Observable.fromEmitter(new Action1<Emitter<GetRoomModel>>() {
    @Override
    public void call(final Emitter<GetRoomModel> emitter) {
        AsyncFunction yourAsyncFunction = /* ... */
        final AsyncFunction.Listener listener = new AsyncFunction.Listener() {
            @Override
            public void onResult(RoomInfo roomInfo) {
                emitter.onNext(roomInfo);
            }
        };
        yourAsyncFunc.doAsync(/* data */, listener);
        emitter.setCancellation(new Cancellable() {
            @Override
            public void cancel() throws Exception {
                // cancel your async call
            }
        });
    }
}, Emitter.BackpressureMode.BUFFER);

您可以对每个异步调用执行相同的操作，然后将它们链接起来。

【讨论】：

感谢您的回复。我会尝试并给你一个反馈。

【解决方案2】：

您不能期望在没有任何代码库的情况下为您提供有关您的应用可以做什么的正确源代码。我建议你在开始编码之前尝试阅读/理解观察者模式

在这里你可以看到观察者模式的结构是如何工作的

          Observable.just(getRoomInfo())
                  .map(info -> sdkApi(info))
                  .map(response -> joinChannel(response))
                  .subscribeOn(Schedulers.newThread())//-->This make your pipeline async.
                  .subscribe(res -> updateUI(res))

在这里您可以看到一些有关 RxJava 工作原理的文档和实际示例。 https://github.com/politrons/reactive

【讨论】：

感谢回复，但是map操作符是一个同步过程，当我将泛型类型设置为int(room id)时，以下步骤的api没有使用int参数。所以我想得到一些关于选择运营商的建议。
Ups 我忘了添加 subscribeOn，现在管道是异步的