java - RxJava Android : Observer incompatible types error

Question

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

Incompatible types

Required: org.reactivestreams.Subscription

Found: void

这是为什么呢？谁能告诉我如何让这个 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;
        }
    });
}

我所有的响应式(Reactive)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

尝试使用一次性复合 Material 。它会是这样的:

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
        }
   }));

不要忘记在 Activity 的 onDestroy 方法上执行compositeDisposable.dispose()。

请告诉我它是否适合您!