[Android] The initial chapter of RxJava

About RxJava

RxJava is an asynchronous operation library launched by ReactiveX for use in the Java VM environment. In addition to the Java environment, ReactiveX also launches Rx libraries for other programming languages, such as Py, Js, Go, etc. There are many introductions and uses of RxJava on the Internet, and there are also many projects using RxJava in Android development. So why use RxJava? Android development also provides asynchronous operation methods for developers to use. I think RxJava is simpler and more elegant than Handle and AsyncTask.

1 RxJava uses chain calls, which is clear and concise in program logic

2 It adopts the extended observer design pattern

I won’t repeat the introduction of the observer pattern and other RxJava. The following content mainly focuses on the use of RxJava and RxAndroid . The RxJava official documentation has been introduced in detail. This section is mainly for learning and discussion. If there are any errors, I hope you can point them out.

Observed Observable

Using RxJava you need to create an Observable, which is used to emit data. The create method of the following Observable needs to pass in an OnSubscribe, which inherits from Action1>. The Subscriber in the Action is the subscriber.

public static <T> Observable<T> create(OnSubscribe<T> f) { 
    return new Observable<T>(RxJavaHooks.onCreate(f)); 
}

In addition, the create method needs to implement the interface call and return the subscriber object. The call method implements the event stream to be executed after the observable is subscribed. subscriber.onNext emits data, and subscriber.onCompleted can indicate the end of the emission event. Then call the observable's subscribe method to implement the event stream that is executed after being subscribed.

Observable<String> observable = Observable 
                .create(new Observable.OnSubscribe<String>() { 
 
            @Override 
            public void call(Subscriber<? super String> subscriber) { 
                subscriber.onNext("1"); 
                subscriber.onNext("2"); 
                subscriber.onNext("3"); 
                subscriber.onNext("4"); 
                subscriber.onNext("5"); 
            } 
}); 
Subscriber<String> subscriber = new  Subscriber<String>() { 
            @Override 
            public void onCompleted() { 
 
            } 
 
            @Override 
            public void onError(Throwable e) { 
 
            } 
 
            @Override 
            public void onNext(String s) { 
                System.out.print(s + &#39;\n&#39;); 
            } 
}; 
observable.subscribe(subscriber); 
//输出结果 print： 
//1 
//2 
//3 
//4 
//5

In addition to using the create method to create an Observable, you can also use from or just to quickly set up the emitted event stream, simplifying the create steps.

Observable<String> o = Observable.from("a", "b", "c");

Observable<String> o = Observable.just("one object");

The asynchronous operation

The thread of RxJava is controlled by the Schedulers scheduler, which controls the thread in which the specific operation is performed.

Schedulers.immediate() Execute in the current thread

Schedulers.newThread() Create a thread for each task to execute

Schedulers.computation() Compute the thread on which the task runs

Schedulers.io() The thread on which the IO task runs ....

AndroidSchedulers.mainThread() Android main thread runs

Thread control is mainly controlled by the two methods subscribeOn() and observeOn():

subscribeOn controls the thread where Observable.OnSubscribe is located, which is equivalent to Observable The thread in which create, just, and from are located.

observeOn controls the thread of the Subscriber, which can also be said to be the thread where the control event is executed.

Observable 
        .just(1,2,3) 
        .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Subscriber<Integer>() { 
            @Override 
            public void onCompleted() { 
 
            } 
 
            @Override 
            public void onError(Throwable e) { 
 
            } 
 
            @Override 
            public void onNext(Integer integer) { 
                 System.out.print(integer + &#39;\n&#39;);                        
            } 
}); 
//输出结果 print： 
//1 
//2 
//3

Write the above RxJava chain call code. Do you feel that it is much more refreshing than the asynchronous call used before? I also said that this is very healing for Virgos!

Operators

ReactiveX provides a lot of operators. Each operator has different functions, but the purpose is to transform and modify the emitted event stream between Observable and Subscribe. This section introduces several common and simple operators. Later, when I have the opportunity, I will write another section on operators to explain the role of each operator in detail. Attached to the official operator documentation, you will know how many there are.

Map()

public final <R> Observable<R> map(Func1<? super T, ? extends R> func) { 
        return create(new OnSubscribeMap<T, R>(this, func)); 
    }

First introduce an operator map. map implements the Func1 interface to transform T type data into R type data and returns R type data. For example, an event queue of type Integer is passed in and returned as String type after map processing.

Observable 
                .just(1,2,3) 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()) 
                .map(new Func1<Integer, String>() { 
                    @Override 
                    public String call(Integer integer) { 
                        return integer + ""; 
                    } 
                }) 
                .subscribe(new Subscriber<String>() { 
                    ...... 
                    @Override 
                    public void onNext(String str) { 
                        System.out.print(str + &#39;\n&#39;); 
                    } 
                }); 
//输出结果 print： 
//1 
//2 
//3

Filter()

public final Observable<T> filter(Func1<? super T, Boolean> predicate) { 
       return create(new OnSubscribeFilter<T>(this, predicate)); 
   }

filter implements the Func1 interface like map, but its converted type is boolean, which filters the emitted event stream. When the converted boolean value is true, the subscriber can receive it Events that pass the filtering, otherwise the event will not be consumed. For example, event stream filtering requires that the int value can be divisible by 2 before it can continue to be delivered, so the final events that the subscriber can consume are 2, 4, 6, 8, and 10.

Observable 
                .just(1,2,3,4,5,6,7,8,9,10) 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()) 
                .filter(new Func1<Integer, Boolean>() { 
                    @Override 
                    public Boolean call(Integer integer) { 
                        return integer % 2 == 0; 
                    } 
                }) 
                .map(new Func1<Integer, String>() { 
                    @Override 
                    public String call(Integer integer) { 
                        return integer + ""; 
                    } 
                }) 
                .subscribe(new Subscriber<String>() { 
                   ...... 
                    @Override 
                    public void onNext(String str) { 
                        System.out.print(str + &#39;\n&#39;); 
                        Log.i("subscribe", str); 
                    } 
                }); 
//输出结果 print： 
//2 
//3 
//4 
//6 
//8 
//10

Skip()

public final Observable<T> skip(int count) { 
        return lift(new OperatorSkip<T>(count)); 
    }

The skip operator means to skip the first few events and start emitting events from a certain event, and the subscript starts from 0.

Observable 
                .just(1,2,3,4,5,6,7,8,9,10) 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()) 
                .skip(3) 
                .map(new Func1<Integer, String>() { 
                    @Override 
                    public String call(Integer integer) { 
                        return integer + ""; 
                    } 
                }) 
                .subscribe(new Subscriber<String>() { 
                    ...... 
                    @Override 
                    public void onNext(String s) { 
                        System.out.print(s + &#39;\n&#39;); 
                        Log.i("subscribe", s); 
                    } 
                }); 
//输出结果 print： 
//4 
//5 
//6 
//7 
//8 
//9 
//10

Range()

public static Observable<Integer> range(int start, int count) { 
        if (count < 0) { 
            throw new IllegalArgumentException("Count can not be negative"); 
        } 
        if (count == 0) { 
            return Observable.empty(); 
        } 
        if (start > Integer.MAX_VALUE - count + 1) { 
            throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE"); 
        } 
        if(count == 1) { 
            return Observable.just(start); 
        } 
        return Observable.create(new OnSubscribeRange(start, start + (count - 1))); 
    }

range operator can be understood as just, from passes a continuous int type array to be emitted, n is the starting int value, m is Count. For example, n = 1, m = 5 int array is {1, 2, 3, 4, 5}

End

I’ve learned this first in this section, which is my initial understanding and learning of RxJava. In fact, using RxJava mainly relies on the use of operators. The operators introduced earlier are the simplest and most basic, and there are many particularly useful operators that have not been introduced. We will continue to introduce some operators later. RxJava is very popular in Android development because of its power. At the same time, RxJava can be used in combination with Retrofit to process network request return values ​​more efficiently. In addition, the android-architecture on GitHub GoogleSample also has TODO projects using the RxJava framework. You can take a look and understand the practical application of RxJava in the project.