java - RxJava中如何同步异步方法？ RxJava 中的异步瀑布

Question

使用RxJava，如何对数据列表顺序执行异步方法？

使用 Node.js Async 模块可以按以下方式顺序调用对象数组上的函数:

var dataArray = ['file1','file2','file3']; // array of arguments
var processor = function(filePath, callback) { // function to call on dataArray's arguments
    fs.access(filePath, function(err) { // perform an async operation
        // process next item in dataArray only after the following line is called
        callback(null, !err)
    }
};
async.every(dataArray, processor, function(err, result) {
    // process results
});

这样做的好处是，在处理器函数中执行的代码可以是异步的，并且回调只能在异步任务完成后运行。这意味着 dataArray 中的每个对象将被依次处理，而不是并行处理。

现在，在 RxJava 中，我可以通过以下方式调用 dataArray 上的处理函数:

String[] dataArray = new String[] {"file1", "file2", "file3"};
Observable.fromArray(dataArray).subscribe(new Consumer<String>() { // in RxJava 1 it's Action1
    @Override
    public void accept(@NonNull String filePath) throws Exception {
        //perform operation on filePath
    }
});

但是，如果我对filePath执行异步操作，如何保证dataArray项的顺序执行？我正在寻找的是这样的东西:

String[] dataArray = new String[] {"file1", "file2", "file3"};
Observable.fromArray(dataArray).subscribe(new Consumer<String>() {
     // process next item in dataArray only after the callback is called
    @Override
    public void accept(@NonNull String filePath, CallNext callback) throws Exception {
         // SomeDatabase will call callback once
         // the asynchronous someAsyncOperation finishes
         SomeDatabase.someAsyncOperation(filePath, callback);
    }
});

此外，一旦 dataArray 的所有项目都已处理完毕，我如何调用一些代码？有点像“完成监听器”？

注意:我意识到我可能弄错了 RX 概念。我这么问是因为我找不到任何关于如何使用 RxJava 实现我上面提到的 Node.js 异步模式的指南。另外，我正在使用 Android，因此没有 lambda 函数。

Answer 1

经过几天的反复试验，我针对这个问题使用的解决方案如下。欢迎提出任何改进建议!

private Observable<File> makeObservable(final String filePath){
    return Observable.create(new ObservableOnSubscribe<File>() {
        @Override
        public void subscribe(final ObservableEmitter<File> e) throws Exception {
            // someAsyncOperation will call the Callback.onResult after
            // having finished the asynchronous operation.
            // Callback<File> is an abstract code I put together for this example
            SomeDatabase.someAsyncOperation(filePath, new Callback<File>(){
                @Override
                public void onResult(Error error, File file){
                    if (error != null){
                        e.onError(error);
                    } else {
                        e.onNext(file);
                        e.onComplete();
                    }
                }
            })
        }
    });
}

private void processFilePaths(ArrayList<String> dataArray){
    ArrayList<Observable<File>> observables = new ArrayList<>();
    for (String filePath : dataArray){
        observables.add(makeObservable(filePath));
    }
    Observable.concat(observables)
            .toList()
            .subscribe(new Consumer<List<File>>() {
                @Override
                public void accept(@NonNull List<File> files) throws Exception {
                    // process results.
                    // Files are now sequentially processed using the asynchronous code.
                }
            });
}

简而言之，发生了什么:

1. 将 dataArray 转换为 Observable 的数组
2. 每个 Observable 在订阅时执行异步操作，并在异步操作完成后将数据提供给 onNext
3. 在 Observable 数组上使用 Observable.concat() 以确保顺序执行
4. 使用.toList()将Observable的结果组合到一个List中

虽然此代码满足了我的要求，但由于以下几个原因，我对此解决方案并不完全满意:

1. 我不确定在 Observable.create()...subscribe(){ 中执行异步代码是否是使用 Observable 的正确方法。
2. 我了解到应谨慎使用 Observable.create ( ref docs )
3. 我不确定我想要实现的行为是否需要为 dataArray 中的每个项目创建一个新的 Observable。对我来说，拥有一个能够顺序释放数据的可观察对象似乎更自然 - 但同样，这可能只是我的旧思维方式。
4. 我读过使用 concatMap 应该可以解决这个问题，但我永远不知道如何将其应用到这个示例中，并且发现 concat 就可以解决这个问题。有人愿意解释一下两者之间的区别吗？

我之前也尝试过在 Observable 数组上使用 .zip 函数，虽然我最终成功获得了结果列表，但异步操作是并行执行的，而不是顺序执行的。