RxJava Android：观察者不兼容类型错误

Question

我想使用 RxJava 和 RxAndroid 进行以下异步网络调用。我在项目依赖项中包含了最新的RxJava (3.0.1) 和RxAndroid (3.0.0)，但在Subscription subscription = .... 块中仍然出现以下错误：

不兼容的类型

必需：org.reactivestreams.Subscription

找到：无效

这是为什么呢？谁能告诉我如何让这个 RxJava observable 工作？

感谢任何帮助。

Subscription subscription = getServerResponse(et_id.getText().toString())// error highlighted in this whole block
       .subscribeOn(Schedulers.io())
       .observeOn(AndroidSchedulers.mainThread())
       .subscribe(new Observer<String>() { 
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onNext(String bitmap) {
               // Handle result of network request
            }

            @Override
            public void onError(@NonNull Throwable e) {
               // Update user interface to handle error
            }

            @Override
            public void onComplete() {
               // Update user interface if needed
            }
       });

public Observable<String> getServerResponse(String string) {
    return Observable.fromCallable(new Callable<String>() {
        @Override
        public String call() throws Exception {
            BufferedReader inFromServer = null;
            String response = "";
            Socket clientSocket = null;
            try {
                clientSocket = new Socket(serverHostname, serverPort);
                DataOutputStream outToServer = new DataOutputStream(clientSocket.getOutputStream());
                inFromServer = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
                outToServer.writeBytes(string + "\n");
                response = inFromServer.readLine();
                clientSocket.close();
            } catch (IOException e) {
                e.printStackTrace();
                response = MainActivity.this.getString(R.string.result_serverError);
            }
            return response;
        }
    });
}

我对反应式 Java 的所有导入：

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

Answer 1

尝试使用一次性复合材料。应该是这样的：

private CompositeDisposable compositeDisposable = CompositeDisposable();
compositeDisposable.add(getServerResponse(et_id.getText().toString())
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())
   .subscribe(new Observer<String>() { 
        @Override
        public void onSubscribe(@NonNull Disposable d) {
        }

        @Override
        public void onNext(String bitmap) {
           // Handle result of network request
        }

        @Override
        public void onError(@NonNull Throwable e) {
           // Update user interface to handle error
        }

        @Override
        public void onComplete() {
           // Update user interface if needed
        }
   }));

不要忘记在活动的 onDestroy 方法上执行 CompositeDisposable.dispose()。

让我知道它是否适合你！