The JAVA RxJava

RxJava overview

Basic use of RxJava

RxJava has three basic elements:

An Observable.
Observer (Observer)
Subscribe (subscribe)

First add dependencies to gradle file:

Implementation 'IO. Reactivex. Rxjava3: rxjava: 3.0.0' implementation 'IO. Reactivex. Rxjava3: rxandroid: 3.0.0'Copy the code

The latest version to view website: RxJava: https://github.com/ReactiveX/RxJava RxAndroid: https://github.com/ReactiveX/RxAndroid

Create the observed

Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable { emitter.onNext(1); emitter.onComplete(); }});Copy the code

Creating an observer

Observer Observer = new Observer<Integer> @override public void onSubscribe(@nonnull Dispo log.d (" data ", "onSubscribe"); } @override public void onNext(@nonnull Integer in log. d(" data ", "onNext" + Integer); } @override public void onError(@nonnull Throwable log.d);} @override public void onError(@nonnull Throwable log.d); } @override public void onComplete() {log.d (" data ", "onComplete"); }};Copy the code

To subscribe to

observable.subscribe(observer);
Copy the code

The observer can only send data to the observer after binding the observer by subscribing.

RxJava core ideas

The core idea of RxJava is observer mode and responsive programming.

Observer mode:Observable observedJust like a producer,The Observer observersLike a consumer, a producer throughsubscribeSubscribe to consumers and start sending data to them.
Responsive programming: TheObservable as a starting point.The Observer is the endIn which various operators can be added.

RxJava cooperate to Retrofit

Retrofit itself does not have the ability to send data like OkHttp request network and RxJav, but it does manage to encapsulate these, so projects often use OkHttp and RxJava to request data with Retrofit. Let’s do a demo:

Request API

public interface NetApi {

  @POST("register")
  @FormUrlEncoded
  Observable<RegisterResponse> registerAction(@Body RegisterRequest request);

}
Copy the code

Create Retrofit to work with OkHttp, RxJava, Gson

public class RetrofitUtil { public static <T> T create(Class<T> tClass) { return new Retrofit.Builder() .baseUrl("http://xxxxxxx") .client(getHttpClient()) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) .build() .create(tClass); } private static OkHttpClient getHttpClient() { OkHttpClient.Builder builder = new OkHttpClient.Builder() .readTimeout(10, TimeUnit.SECONDS) .connectTimeout(9, TimeUnit.SECONDS); if (BuildConfig.DEBUG) { HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor(); interceptor.setLevel(HttpLoggingInterceptor.Level.BODY); builder.addInterceptor(interceptor); } return builder.build(); }}Copy the code

RxJava gets the data flow

RetrofitUtil.create(NetApi.class)
      .registerAction(new RegisterRequest())
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Observer<RegisterResponse>() {
          @Override
          public void onSubscribe(@NonNull Disposable d) {
          }
          @Override
          public void onNext(@NonNull RegisterResponse registerRe
          }
          @Override
          public void onError(@NonNull Throwable e) {
          }
          @Override
          public void onComplete() {
          }
      });
Copy the code

With Retrofit, we don’t have to write OkHttp request network execution code or convert Json strings to objects. Just define RxJava operations and receive.

RxJava patterns and principles

RxJava operator application

Create, just, fromIterable, etc. The map operator is used to manipulate the data sent by the transformation. Recreate Observable sending data with the flatMap operator; Send multiple events combined using the concat and merge operators. The threading environment for the above code is determined by the subscribeOn operator and that for the following code by the observeOn operator. More use of manipulation symbols can be referenced: RxJava2 this article is enough

RxJava create principle analysis

The above code can be divided into two parts and is also the simplestStarting and ending points.

1. Observables. The create () returnsObservableCreateclass

@CheckReturnValue
@NonNull
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(@NonNull ObservableOnSubscribe<T> source) {
   Objects.requireNonNull(source, "source is null");
   return RxJavaPlugins.onAssembly(new ObservableCreate<>(source));
}
Copy the code

And pass the ObservableOnSubscribe as the source argument to the ObservableCreate class:

Public Final Class ObservableCreate<T> extends Observable<T> {// Insert 'ObservableOnSubscribe' into the final ObservableOnSubscribe<T> source; public ObservableCreate(ObservableOnSubscribe<T> source) { this.source = source; }...Copy the code

2. BysubscribeSubscribe to the observer, the execution ObservableCreate. The subscribe () method:

The ObservableCreate executes the.subscribe() method of the parent ObservableCreate, and finally theObserverInstance is passed to the ObservableCreate classsubscribeActualMethod execution:

We first wrapped the observer in a layer and called the observer.onsubscribe () method
Subscribe to the source interface instance and pass the parent package.

3. Execute the data to the Observer

The subscribe method of the instance of the source