Written in Jane’s book

Hi ~ I am RxJava

RxJava is now available in versions 1.0 and 2.0. The content described here is primarily based on version 1.0, with a slight reference to 2.0.

introduce

A library that uses observable sequences to compose asynchronous, event-based programs on the Java VM.

Project home page

reactivex.io/

Chinese document

McXiaoke. Gitbooks. IO/rxdocs/cont…

The API documentation

Reactivex. IO/RxJava/Java…

The project address

Github.com/ReactiveX/R… Github.com/ReactiveX/R…

The basic concept

0. Basic mode

1.Observable, an event source.

The object being observed in observer mode. Be upstream of the process.

2.Observer

The observer, the receiver. In observer mode observer. The following methods need to be implemented:

  • onNext(T item)

The Observable calls this method to emit data, and its parameters are the data that the Observable emits. This method can be called multiple times, depending on your implementation.

  • onError(Exception ex)

This method is called when an Observable encounters an error or fails to return the expected data. The call terminates the Observable, and onNext and onCompleted are not called. The arguments to the onError method are thrown exceptions.

  • onComplete

Observable calls this method after the last onNext call if no errors are encountered.

3.Subscriber

The subscriber is also the recipient. There is one more Subscription than the Observer.

 public abstract class Subscriber<T> implements Observer<T>, Subscription {
 ......
 }
Copy the code
public interface Subscription {
    /**
     * Stops the receipt of notifications on the {@link Subscriber} that was registered when this Subscription
     * was received.
     * <p>
     * This allows unregistering an {@link Subscriber} before it has finished receiving all events (i.e. before
     * onCompleted is called).
     */
    void unsubscribe();

    /**
     * Indicates whether this {@code Subscription} is currently unsubscribed.
     *
     * @return {@code true} if this {@code Subscription} is currently unsubscribed, {@code false} otherwise
     */
    boolean isUnsubscribed();
}
Copy the code
  • unsubscribe();

unsubscribe

  • isUnsubscribed();

Determine whether to subscribe

The result of unsubscribing is passed to the Observable operator chain and causes every link in the chain to stop emitting data items. This is not guaranteed to happen immediately, however, for an Observable to continue generating and attempting to emit items in a while loop even when there are no observers left.

4. Operator: Operator

There are several main requirements

Create an Observable directly.

  • Combine multiple Observables
  • Perform transformations on data emitted by an Observable (transform)
  • Retrieves specific values from data emitted by an Observable (filtering operation)
  • Forwarding partial Observable values (conditional/Boolean/filter operations)
  • Evaluate data sequences emitted by an Observable (arithmetic/aggregate operations)

If you want to implement your own operators, see here: Implementing custom operators

The Scheduler can switch between the main thread and various threads.

  • Schedulers.immediate()

Running in the current thread is equivalent to not switching threads. This is the default Scheduler.

  • Schedulers.newThread()

Always enable a new thread and perform operations on the new thread.

  • Schedulers.io()

Scheduler used for I/O operations (reading and writing files, databases, network information interaction, and so on). The behavior of IO () is similar to that of newThread(), except that the internal implementation of IO () uses an unlimited pool of threads and can reuse idle threads, so in most cases IO () is more efficient than newThread(). Don’t put your calculations in IO () to avoid creating unnecessary threads.

  • Schedulers.computation()

The Scheduler used for the calculation. This calculation refers to CPU intensive computing, that is, operations that do not limit performance by operations such as I/O, such as graphics computation. This Scheduler uses a fixed thread pool of CPU cores. Don’t focus on COMPUTATION (), or WAIT time for I/O operations will waste CPU.

  • AndroidSchedulers.mainThread()

Switch to the main thread, and the specified operation will run on the Android main thread.

ActionX: Function with no return value

Action0.class

public interface Action0 extends Action {
    void call();
}
Copy the code

Action1.class

public interface Action1<T> extends Action {
    void call(T t);
}

Copy the code

Action2.class

public interface Action2<T1, T2> extends Action {
    void call(T1 t1, T2 t2);
}

Copy the code

FuncX: a function that returns a value

Func0.class

public interface Func1<T, R> extends Function {
    R call(T t);
}
Copy the code

Func1.class

public interface Func1<T, R> extends Function {
    R call(T t);
}
Copy the code

Func2.class

public interface Func2<T1, T2, R> extends Function {
    R call(T1 t1, T2 t2);
}
Copy the code

The difference from Action is whether an argument is returned.

The operator

The create operation

Operator used to create an Observable

  • Create: Creates an Observable from scratch by calling the observer method
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { subscriber.onNext("Observable Create call"); subscriber.onCompleted(); }});Copy the code
  • Empty: Creates a special Observable with limited behavior
 Observable deferObservable = Observable.empty();
Copy the code
  • From: Converts other objects or data structures into Observables
List<String> list = new ArrayList<>(); list.add("from1"); list.add("from2"); list.add("from3"); Observable fromObservable = Observable.from(list); // Iterate through the list one at a timeCopy the code

Transform operation

These operators can be used to transform data emitted by an Observable, and can be explained in the documentation for each operator

  • Buffer: A cache that collects data from observables into a collection on a regular basis, and then packs the collection instead of sending it one at a time
Observable rangeObservable = Observable.from(list).buffer(3);
Copy the code
  • Map: Transforms the data emitted by an Observable by applying a function to each item in the sequence. Essentially, it executes a function on each item in the sequence. The parameter of the function is the data item
Observable observable = Observable.from(list).map(new Func1<String, String>() { @Override public String call(String string) { return string + " map"; }});Copy the code

Filtering operation

These operators are used to select from the data emitted by an Observable

  • Filter: to filter out data items that do not pass the predicate test and only fire those that do
Observable.just(1, 2, 3, 4, 5, 6)
                .filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer>4;
                    }
                }).subscribe(ObserverFactory.createIntObserver());
Copy the code
  • First: the first item, only the first data that meets the conditions is transmitted
Observable.just(1, 2, 3, 4, 5, 6)
                .first(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer>2;
                    }
                }).subscribe(ObserverFactory.createIntObserver());
Copy the code

Arithmetic and aggregation operations

These operators can be used for entire data sequences Concat: connecting data from multiple Observables without interleaving them

Observable<Integer> source1 =  Observable.just(1, 2, 5, 6);
        Observable<Integer> source2 =  Observable.just(7, 6, 5, 6);
        Observable.concat(source1,source2).distinct().subscribe(ObserverFactory.createIntObserver());
Copy the code

Auxiliary operation

A set of operators that handle Observables

  • Delay: indicates the delay of transmitting result data
Observable justObservable = Observable.just("just1","just2"); / / in order to send "just1" and "just2" justObservable. Delay (3000, TimeUnit. MILLISECONDS). The subscribe (ObserverFactory. CreateObserver ());Copy the code
  • ObserveOn: a scheduler (worker thread) that specifies an observer to observe an Observable
Observable justObservable = Observable.just("just1","just2"); / / in order to send "just1" and "just2" justObservable. ObserveOn (Schedulers. Immediate ()). The subscribe (ObserverFactory. CreateObserver ());Copy the code

More operator examples

Github.com/iamludaxu/a…

Other related content

RxJava by parabola for Android developers
The friendliest article about RxJava
Most friendly articles on RxJava (Advanced)
RxJava all operator examples
RxJava2.0 tutorial series for beginners


My name is Lu Daxu.

A boring programmer who knows something about psychology. Tip, follow and like the article whether or not you get anything from it!