Android-RxJava: 使用observable从后台线程更新UI

Question

我刚开始构建一个简单的Android应用程序，希望在后台线程中进行网络请求，然后使用服务器响应更新主线程（UI线程）。到目前为止，我使用了AsyncTasks，但是未来的实现我想使用reactive Java (RxJava)。我以前没有做过响应式调用，所以想要一个简单但完整的示例（Observable和Observer的创建和订阅），以便可以进一步构建。

我已经将RxJava依赖项包含到基本的Android项目中，并编写了一个使用AsyncTasks进行网络请求的非常简单的主活动。现在我尝试用响应式方式替换AsyncTask实现，但卡在有关Observable和Observer的所有信息上。我不确定什么是最小但完全可工作的示例所必需的。

我真的很感谢您能帮助我将主要部分转换为反应式实现，因为我不知道如何处理从响应字符串生成Observable并订阅Observer。

谢谢。

package com.example.reactiveTest;

import androidx.appcompat.app.AppCompatActivity;

import android.os.AsyncTask;
import android.os.Bundle;
import android.view.View;
import android.widget.Button;
import android.widget.TextView;

public class MainActivity extends AppCompatActivity {

    private Button btnSend = null;
    private TextView result = null;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        this.btnSend = findViewById(R.id.button_send);
        this.result = findViewById(R.id.result);
    }

    public void onClickBtnSend(View view) {
        new SomeTask().execute("Just some String");
    }

    class SomeTask extends AsyncTask<String, Void, String> {

        @Override
        protected String doInBackground(String... strings) {
            // server request returning response String
            return response;
        }

        @Override
        protected void onPostExecute(String string) {
            // update UI with response String
            result.setText(string);
        }
    }
}

- avermaet

---

这篇简短的文章应该能帮助你从asynctask迁移到rx。 - Husayn Hakeem

@HusaynHakeem 谢谢，但是和 onError、onNext 和 onComplete 方法有什么区别呢？它们到底是用来做什么的？ - avermaet

Answer 1

使用AsyncTask，你基本上是在工作线程上执行异步操作，然后在主线程上使用其结果。在Rx中，你可以使用以下代码：

Observable.fromCallable(asyncOperation)
   .subscribeOn(backgroundThread)
   .observeOn(mainThread)
   .subscribe(result -> { /* update UI for instance */ })

看起来你也对onNext、onError和onComplete感兴趣。

• onNext在observable每次发出一个项目时都会被调用。每次调用时，它都会接收一个项目，然后可以处理它。
• onError在observable因任何原因遇到错误时被调用。当它被调用时，它将接收一个Throwable，表示错误的原因。在它被调用后，onNext和onComplete不会被调用。
• onComplete在onNext用最后一个项目调用后被调用。它不接收任何输入，例如，可以在其中进行一些清理操作。

使用上述方法的样子如下：

Observable.fromCallable(asyncOperation)
   .subscribeOn(backgroundThread)
   .observeOn(mainThread)
   .subscribe(onNext, onError, onComplete)

如果你想使用 Observable.create() 来创建 Observable，那是完全可以的，它可以让你更好地控制 Observable 何时以及如何发射数据。例如，如果你想处理网络请求中可能出现的某些特定错误，并根据错误类型发射不同的 Throwable，你就可以这样做。

ObservableOnSubscribe asyncOperation = new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> emitter) {
        try {
            // network request
            // Once result is ready, call emitter.onNext().
            // When done, complete this Observable by calling emitter.onComplete()
        } catch (Exception e) {
            // handle error, and emit it using emitter.onError()
        }
    }
 } 

- Husayn Hakeem

---

如果使用Observable.create(...)而不是fromCallable()创建Observable会怎样呢？ - avermaet