RxJava2同步多文件上传：获取单文件回调更新进度对话框

嘿，作为RxJava新手碰到这个进度跟踪的问题太正常了！我来帮你梳理下怎么解决——其实你不用纠结zip操作符，它更适合等待所有流完成后合并结果，完全满足不了你实时跟踪每个文件进度的需求。换个思路，我们为每个文件单独创建上传流，再组合这些流，就能轻松拿到每个文件的onNext和onCompleted回调啦！

核心思路 #

1. 为每个文件封装独立的上传Flowable，在这个流内部处理单个文件的进度更新和完成/错误回调
2. 用concat（串行上传）或merge（并行上传）把所有单个文件的上传流组合起来
3. 订阅组合后的流，处理全局的完成/错误逻辑

具体代码实现 #

第一步：封装单个文件上传的Flowable

先把你原来的单文件上传逻辑封装成可复用的方法，这里我们加入进度跟踪和单个文件的完成回调：

private Flowable<FileLinkProgress> uploadSingleFile(SelectablePhoto photo) {
    String filePath = photo.getPhotoModel().getPath();
    if (filePath == null) {
        return Flowable.error(new IllegalArgumentException("文件路径为空"));
    }

    Config config = new Config(FILE_STACK_API_KEY);
    Client client = new Client(config);

    return client.uploadAsync(filePath, true)
            .subscribeOn(Schedulers.io())
            // 处理单个文件的进度/完成数据
            .doOnNext(fileLinkProgress -> {
                AndroidSchedulers.mainThread().scheduleDirect(() -> {
                    if (fileLinkProgress.getData() != null) {
                        // 文件上传完成：更新本地数据+UI提示
                        String remotePath = fileLinkProgress.getData().imageTransform().url();
                        photo.getPhotoModel().setPath(remotePath);
                        uploadDialog.setMessage("已完成：" + photo.getPhotoModel().getFileName());
                    } else {
                        // 上传中：如果API返回进度百分比，这里可以实时更新
                        // 假设fileLinkProgress提供getProgress()方法获取进度
                        int progress = fileLinkProgress.getProgress();
                        uploadDialog.setMessage("正在上传：" + photo.getPhotoModel().getFileName() + " (" + progress + "%)");
                    }
                });
            })
            // 单个文件上传完成的回调
            .doOnComplete(() -> {
                Log.d("Upload", "文件上传完成：" + photo.getPhotoModel().getFileName());
                AndroidSchedulers.mainThread().scheduleDirect(() -> {
                    Toast.makeText(getContext(), photo.getPhotoModel().getFileName() + " 上传成功", Toast.LENGTH_SHORT).show();
                    // 如果用进度条，可以递增进度
                    // uploadDialog.incrementProgressBy(1);
                });
            })
            // 单个文件上传失败的回调
            .doOnError(throwable -> {
                Log.e("Upload Error", "上传失败：" + photo.getPhotoModel().getFileName(), throwable);
                AndroidSchedulers.mainThread().scheduleDirect(() -> {
                    uploadDialog.setMessage("上传失败：" + photo.getPhotoModel().getFileName());
                    // 这里可以选择是否继续上传其他文件，比如忽略错误继续
                });
            });
}

第二步：组合多文件上传流并订阅

接下来把所有要上传的文件转换成Flowable列表，再组合起来订阅：

// 获取用户选择的所有文件
List<SelectablePhoto> allPhotos = currentProduct.getSelectablePhotos();

// 初始化ProgressDialog（设置总进度为文件数量）
uploadDialog = new ProgressDialog(getContext());
uploadDialog.setProgressStyle(ProgressDialog.STYLE_HORIZONTAL);
uploadDialog.setMax(allPhotos.size());
uploadDialog.setMessage("准备上传...");
uploadDialog.show();

// 创建每个文件的上传Flowable列表
List<Flowable<FileLinkProgress>> uploadFlowables = new ArrayList<>();
for (SelectablePhoto photo : allPhotos) {
    uploadFlowables.add(uploadSingleFile(photo));
}

// 组合流：用concat串行上传（按顺序），用merge并行上传（同时）
Flowable.concat(uploadFlowables)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
                // 这里的onNext会触发每个文件的上传事件，但我们已经在doOnNext里处理了
                fileLinkProgress -> {},
                // 全局错误处理：如果某个流抛出致命错误会触发这里
                throwable -> {
                    Log.e("Upload Error", "全局上传错误", throwable);
                    uploadDialog.dismiss();
                    Toast.makeText(getContext(), "上传过程中出现异常", Toast.LENGTH_LONG).show();
                },
                // 所有文件上传完成的回调
                () -> {
                    uploadDialog.dismiss();
                    Toast.makeText(getContext(), "所有文件上传完成！", Toast.LENGTH_LONG).show();
                    // 这里可以做后续操作，比如提交产品数据到服务器
                }
        );

关键细节说明 #

• 为什么不用zip？：zip会等待所有传入的Flowable都完成后，才一次性发射组合后的结果，完全没法实时跟踪每个文件的进度和完成状态，对你的场景不适用。
• concat vs merge：
  • concat：按顺序逐个上传文件，适合对上传顺序有要求的场景
  • merge：并行上传多个文件，速度更快，但要注意UI更新的并发问题，以及API是否允许并行上传
• 线程切换：所有UI操作必须切换到主线程，这里用AndroidSchedulers.mainThread().scheduleDirect()确保安全更新UI
• 错误处理：doOnError处理单个文件的上传失败，全局的subscribe错误回调处理致命异常，你可以根据需求决定是否终止整个上传流程

内容的提问来源于stack exchange，提问作者Ofek Pintok