您好，我刚刚开始使用 RxJava2 学习响应式编程。 如何创建一个在后台线程中运行的任务，然后使用 RxJava2 在主线程中完成。

在 Android 中我们使用 AsyncTask 的例子就像下面的例子

private class MyTask extends AsyncTask<String, Integer, Boolean>
{
    @Override
    protected Boolean doInBackground(String... paths)
    {
        for (int index = 0; index < paths.length; index++)
        {
            boolean result = copyFileToExternal(paths[index]);

            if (result == true)
            {
                // update UI
                publishProgress(index);
            }
            else
            {
                // stop the background process
                return false;
            }
        }

        return true;
    }

    @Override
    protected void onProgressUpdate(Integer... values)
    {
        super.onProgressUpdate(values);
        int count = values[0];
        // this will update my textview to show the number of files copied
        myTextView.setText("Total files: " + count);
    }

    @Override
    protected void onPostExecute(Boolean result)
    {
        super.onPostExecute(result);
        if (result)
        {
            // display a success dialog
            ShowSuccessAlertDialog();
        }
        else
        {
            // display a fail dialog
            ShowFailAlertDialog();
        }
    }
}

对于这个例子，我想传入一个 Array/ArrayList 的字符串，它用于在后台线程 中执行一些方法。然后每个成功结果都会更新我的 TextView(UI 线程)。如果其中一个过程失败，我希望它直接停止。最后，我想在流程完成后更新我的 View 。

我只做到了这一点

Observable.just(paths).subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<ArrayList<String>>()
            {
                private boolean result;

                @Override
                public void onSubscribe(Disposable d)
                {
                }

                @Override
                public void onNext(ArrayList<String> paths)
                {
                    for (int index = 0; index < paths.size(); index++)
                    {
                        result = copyFileToExternal(paths[index]);

                        if (result == true)
                        {
                            // cant update UI because is in background thread
                            myTextView.setText("Total files: " + index);
                        }
                        else
                        {
                            // end the loop
                            break;
                        }
                    }
                }

                @Override
                public void onError(Throwable e)
                {
                }

                @Override
                public void onComplete()
                {
                    if (result)
                    {
                        // cant display because it is still in background thread
                        ShowSuccessAlertDialog();
                    }
                    else
                    {
                        // cant display because it is still in background thread
                        ShowFailAlertDialog();
                    }
                }
            });

我看了一些教程，但似乎找不到答案。

在此先感谢您的帮助

最佳答案

我会做这样的事情:

Observable.fromArray(getPaths())
    .map(path -> copyFileToExternal(path))
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(aInteger -> Log.i("test", "update UI"),
               throwable -> ShowFailAlertDialog),
               () -> ShowSuccessAlertDialog());

一个好主意通常是有一个“处理程序”来控制对您的观察者的订阅。这样，当您需要停止后台任务时(例如因为用户离开了 Activity)，您可以使用它。为此，您可以使用 subscribeWith 而不是 subscribe，它接收 ResourceObserver 作为输入:这样您就可以获得 Disposable。

Disposable subscription = Observable.fromArray(getPaths())
    .map(path -> copyFileToExternal(path))
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeWith(new ResourceObserver<Integer>() {
         @Override
         public void onNext(@NonNull Integer index) {
             Log.i("test", "update UI");
         }
         @Override
         public void onError(@NonNull Throwable e) {
             ShowFailAlertDialog();
         }
         @Override
         public void onComplete() {
             ShowSuccessAlertDialog();
         }
   });

当你需要停止任务时，你可以调用:

subscription.dispose();

关于android - RxJava2 类似于 Android 中的 AsyncTask，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43533890/