如何用 Rx 等效项替换 AsyncTask 和 AsyncTaskLoader？

Question

背景

有些文章声称 Rx 可以替代 AsyncTask 和 AsyncTaskLoader。 看到 Rx 通常会使代码更短，我尝试深入研究各种示例和有关其工作原理的想法。

问题

我发现的所有示例和文章，包括 Github 存储库，似乎都不能很好地替代 AsyncTask 和 AsyncTaskLoader：

1. 找不到如何取消一个或多个任务（包括中断线程）。这对于 AsyncTask 尤为重要，通常用于 RecyclerView 和 ListView。这也可以在 AsyncTaskLoader 上完成，但需要比仅仅调用“取消”更多的工作。对于 AsyncTask，它还提供进度的发布，这是我在 Rx 中找不到示例的另一件事。

2. 找不到如何在避免内存泄漏的同时避免手动取消订阅，这完全破坏了使用 Rx 的全部意义，因为它应该更短。在某些示例中，回调的数量甚至比普通代码还要多。

我尝试过的

这是我读过的一些关于 Rx 的链接：

• https://stablekernel.com/replace-asynctask-and-asynctaskloader-with-rx-observable-rxjava-android-patterns/ - 这实际上提供了一种很酷的方式来处理 AsyncTaskLoader，因为它缓存了结果。但是，这些示例根本不起作用（包括构建它们），因为它们非常旧。

• https://code.tutsplus.com/tutorials/getting-started-with-rxjava-20-for-android--cms-28345、https://code.tutsplus.com/tutorials/reactive-programming-operators-in-rxjava-20--cms-28396、https://code.tutsplus.com/tutorials/rxjava-for-android-apps-introducing-rxbinding-and-rxlifecycle--cms-28565 - 所有这些都主要谈论 RxJava，而不是关于替换 AsyncTask 或 AsyncTaskLoader。

• https://github.com/L4Digital/RxLoader - 需要取消订阅，即使它没有说它需要它。

问题

如何使用 Rx 替代 AsyncTask 和 AsyncTaskLoader？

例如，我如何将这个微小的 AsyncTaskLoader 示例代码替换为 Rx 等效代码：

class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
        supportLoaderManager.initLoader(1, Bundle.EMPTY, object : LoaderCallbacks<Int> {
            override fun onCreateLoader(id: Int, args: Bundle): Loader<Int> {
                Log.d("AppLog", "onCreateLoader")
                return MyLoader(this@MainActivity)
            }

            override fun onLoadFinished(loader: Loader<Int>, data: Int?) {
                Log.d("AppLog", "done:" + data!!)
            }

            override fun onLoaderReset(loader: Loader<Int>) {}
        })
    }

    private class MyLoader(context: Context) : AsyncTaskLoader<Int>(context) {

        override fun onStartLoading() {
            super.onStartLoading()
            forceLoad()
        }

        override fun loadInBackground(): Int {
            Log.d("AppLog", "loadInBackground")
            try {
                Thread.sleep(10000)
            } catch (e: InterruptedException) {
                e.printStackTrace()
            }
            return 123
        }
    }
}

对于 AsyncTask，通常我们为每个 ViewHolder 设置它，作为一个字段，然后在 onBindViewHolder 以及活动/片段被销毁时取消它（检查所有仍处于挂起或运行的状态）。

类似这样的东西（样本尽可能短，当然应该根据需要进行更改）：

class AsyncTaskActivity : AppCompatActivity() {
    internal val mTasks = HashSet<AsyncTask<Void, Int, Void>>()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_async_task)
        val recyclerView = findViewById<RecyclerView>(R.id.recyclerView)
        recyclerView.layoutManager = LinearLayoutManager(this, LinearLayoutManager.VERTICAL, false)
        recyclerView.adapter = object : Adapter<MyViewHolder>() {
            override fun onCreateViewHolder(parent: ViewGroup, viewType: Int): MyViewHolder {
                return MyViewHolder(LayoutInflater.from(this@AsyncTaskActivity).inflate(android.R.layout.simple_list_item_1, parent, false))
            }

            @SuppressLint("StaticFieldLeak")
            override fun onBindViewHolder(holder: MyViewHolder, position: Int) {
                holder.tv.text = "loading..."
                if (holder.task != null) {
                    holder.task!!.cancel(true)
                    mTasks.remove(holder.task!!)
                }
                holder.task = object : AsyncTask<Void, Int, Void>() {
                    override fun doInBackground(vararg voids: Void): Void? {
                        for (i in 0..100)
                            try {
                                Thread.sleep(5)
                                publishProgress(i)
                            } catch (e: InterruptedException) {
                                e.printStackTrace()
                            }

                        return null
                    }

                    override fun onProgressUpdate(vararg values: Int?) {
                        super.onProgressUpdate(*values)
                        holder.tv.text = "progress:" + values[0]
                    }

                    override fun onPostExecute(aVoid: Void?) {
                        super.onPostExecute(aVoid)
                        mTasks.remove(holder.task)
                        holder.tv.text = "done:" + position
                        holder.task = null
                    }
                }.execute()
            }

            override fun getItemCount(): Int {
                return 1000
            }
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        for (task in mTasks)
            task.cancel(true)
    }

    private class MyViewHolder(itemView: View) : ViewHolder(itemView) {
        internal var tv: TextView
        internal var task: AsyncTask<Void, Int, Void>? = null

        init {
            tv = itemView.findViewById(android.R.id.text1)
        }
    }
}

Answer 1

让我提前告诉你，我在 RxJava 方面的经验并不像某些人所说的那样提前。我只知道使用 RxJava + RxAndroid 替换 AsyncTask。例如，让我们使用 Observable 将您的 AsyncTask 代码转换为 RxJava2 等价物，这可以说是最流行的流类型。

Observable
    .create <Int> { emitter ->
        try { 
            for (i in 0..100) {
                Thread.sleep(5) 
                emitter.onNext(i) 
            }
        } catch (e: InterruptedException) {
            emitter.onError(e)
        }
        emitter.onComplete()
    ‎}
    ‎.subscribeOn(Schedulers.io())
    ‎.observeOn(AndroidSchedulers.mainThread())
    ‎.subscribe({ i ->
    ‎    holder.tv.text = "progress:" + i
     }, { e ->
     ‎   e.printStackTrace()
     ‎}, {
     ‎   holder.tv.text = "done:" + position
     ‎})

现在让我们逐行浏览代码：

1. create 允许您自定义发射器行为。请注意，这不是创建流的唯一方法，还有just、fromArray、fromList、defer 等。如果您仍有兴趣，我们可以讨论更多。
2. subscribeOn(Schedulers.io()) 告诉 RxJava #1 将在 I/O 线程中执行。如果这是后台操作，Schedulers.newThread() 也可以工作（至少据我所知）。
3. observeOn(AndroidSchedulers.mainThread()) 告诉 RxJava #4 将在 Android 的主线程中执行。这也是 RxAndroid 填补空白的地方，因为 RxJava 没有任何 Android 参考。
4. subscribe()
   • onNext(i : Int) - 相当于publishProgress()。在较新的 RxJava 中，发出的值不能为空。
   • onError(e : Throwable) - 长期以来被 RxJava 的程序员推崇的错误处理。在较新的 RxJava 中，必须提供 onError()，如果它为空则无关紧要。
   • onComplete() - 相当于onPostExecute()。

同样，我的回答可能有很多错误信息。我希望有更好理解的人来纠正它或提供比这个更好的答案。

Answer 2

似乎没有办法以更短或更简单的形式用RxJava 实现AsyncTask 功能。使用RxJava 可以简化任务链接，因此如果您只有一个任务AsyncTask 可以是更好的解决方案，但是当您有多个连续任务要执行时-RxJava 的界面更方便。因此，例如，您应该为每个任务编写onCancelled，而不是实现一些通用逻辑。