使用RxJava和Okhttp

Question

我想在另一个线程(如IO线程)中使用okhttp请求一个url并进入ResponseAndroid主线程,但我不知道如何创建一个Observable.

Answer 1

首先添加RxAndroid到您的依赖项,然后Observable像这样创建:

 Subscription subscription =   Observable.create(new Observable.OnSubscribe<Response>() {
        OkHttpClient client = new OkHttpClient();
          @Override
          public void call(Subscriber<? super Response> subscriber) {
            try {
              Response response = client.newCall(new Request.Builder().url("your url").build()).execute();
              if (response.isSuccessful()) {
                  if(!subscriber.isUnsubscribed()){
                     subscriber.onNext(response);
                  }
                  subscriber.onCompleted();
              } else if (!response.isSuccessful() && !subscriber.isUnsubscribed()) {
                  subscriber.onError(new Exception("error"));
                }
            } catch (IOException e) {
              if (!subscriber.isUnsubscribed()) {
                  subscriber.onError(e);
              }
            }
          }
        })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<Response>() {
              @Override
              public void onCompleted() {

              }

              @Override
              public void onError(Throwable e) {

              }

              @Override
              public void onNext(Response response) {

              }
            });

它会在另一个线程(io线程)中请求你的url并在android主线程上观察它.

最后当你离开屏幕使用时,subsribtion.unsubscribe()以避免内存泄漏.

当你使用时Observable.create,你应该写很多样板代码,你也必须自己处理订阅.更好的选择是使用延迟.形成文件:

在观察者订阅之前不要创建Observable,并为每个观察者创建一个新的Observable

Defer运算符等待观察者订阅它,然后它生成一个Observable,通常带有Observable工厂函数.它为每个订户重新执行此操作,因此尽管每个订户可能认为它订阅了相同的Observable,但实际上每个订户都获得其自己的单独序列.

正如MarcinKoziński所提到的,你只需要这样做:

final OkHttpClient client = new OkHttpClient();
Observable.defer(new Func0<Observable<Response>>() {
    @Override public Observable<Response> call() {
        try {
            Response response = client.newCall(new Request.Builder().url("your url").build()).execute();
            return Observable.just(response);
        } catch (IOException e) {
            return Observable.error(e);
        }
    }
});

Answer 2

使用它更容易,更安全,Observable.defer()而不是Observable.create():

final OkHttpClient client = new OkHttpClient();
Observable.defer(new Func0<Observable<Response>>() {
    @Override public Observable<Response> call() {
        try {
            Response response = client.newCall(new Request.Builder().url("your url").build()).execute();
            return Observable.just(response);
        } catch (IOException e) {
            return Observable.error(e);
        }
    }
});

这样就可以为您处理取消订阅和背压.这是Dan Lew关于create()和的一篇伟大的文章defer().

如果你希望走的Observable.create()路线,那么它应该看起来更像是在该库与isUnsubscribed()到处洒电话.我相信这仍然无法处理背压.

Answer 3

我意识到这篇文章有点陈旧,但现在有一种新的,更方便的方法

Observable.fromCallable {
        client.newCall(Request.Builder().url("your url").build()).execute()
    }

更多信息:https://artemzin.com/blog/rxjava-defer-execution-of-function-via-fromcallable/