Shulou(Shulou.com)06/02 Report--

This article introduces you how to understand asynchronous programming RxJava, the content is very detailed, interested friends can refer to, hope to be helpful to you.

Preface

Some time ago, I wrote an understanding of collaborative programming, which mentioned that whether it is collaborative programming or callback, it actually provides an asynchronous non-blocking programming mode, and introduces the support for asynchronous non-blocking programming mode in java, mainly referring to Future and CompletableFuture. After that, some students mentioned RxJava in the following message. I happened to be reading the book of microservice design recently, which mentioned responsive extension (Reactive extensions,Rx), and RxJava is the implementation of Rx on JVM, so I intend to learn more about RxJava.

Introduction to RxJava

RxJava's official website address: https://github.com/ReactiveX/RxJava

There is a sentence description of RxJava: RxJava-Reactive Extensions for the JVM-a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

The main idea is: a library that uses observable sequences on Java VM to form asynchronous, event-based programs.

A more detailed explanation describes the main features of RxJava in an article on the Netflix Technology blog:

Easy to concurrency and thus make better use of the capabilities of the server.

It is easy to conditional asynchronous execution.

A better way to avoid callback to hell.

A responsive approach.

Compared with CompletableFuture

It was mentioned earlier that CompletableFuture actually implements the asynchronous programming pattern, a more common usage scenario:

CompletableFuture future = CompletableFuture.supplyAsync (time-consuming function); Future f = future.whenComplete ((v, e)-> {System.out.println (v); System.out.println (e);}); System.out.println ("other...")

Let's use a simple example to see how RxJava implements an asynchronous programming mode:

Bservable observable = Observable.just (1,2) .subscribeOn (Schedulers.io ()) .map (new Func1 () {@ Overridepublic Long call (Integer t) {try {Thread.sleep (1000); / / time-consuming operation} catch (InterruptedException e) {e.printStackTrace ();} return (long) (t * 2);}}) Observable.subscribe (new Subscriber () {@ Overridepublic void onCompleted () {System.out.println ("onCompleted");} @ Overridepublic void onError (Throwable e) {System.out.println ("error" + e);} @ Overridepublic void onNext (Long result) {System.out.println ("result =" + result);}}); System.out.println ("other...")

A time-consuming operation is performed asynchronously in Func1, and the Subscriber (observer) is subscribed to the Observable (observed). When the time-consuming operation is finished, the onNext method in Subscriber will be called back.

The async mode is specified in subscribeOn (Schedulers.io ()), which can be understood as starting a new thread each time a time-consuming operation is performed.

In structure, it is actually very similar to CompletableFuture, which performs a time-consuming operation asynchronously and then proactively tells me the result when there is a result. Then why do we need RxJava? I don't know if you have noticed. The above example actually provides two data streams [1Magne2], and you will actively tell me after dealing with any one of them. Of course, this is only one of its functions. RxJava also has many useful features, which will be introduced in the following content.

Asynchronous observer mode

Does the above code find anything particularly similar to the one in the design pattern: observer pattern; first provide an observer Observable, and then add the observer Subscriber to the observer list

There are four roles available in RxJava: Observable, Observer, Subscriber, and Subjects

Observables and Subjects are the two observers, Observers and Subscribers are the observers.

Of course, we can also take a look at the source code, look at Observer in jdk and Observer in RxJava.

Observer in jdk:

Public interface Observer {void update (Observable o, Object arg);}

Observer of RxJava:

Public interface Observer {void onCompleted (); void onError (Throwable e); void onNext (T t);}

At the same time, it can be found that Subscriber is implements Observer:

Public abstract class Subscriber implements Observer, Subscription

It can be found that RxJava introduces two new methods into Observer: onCompleted () and onError ().

OnCompleted (): notifies the observer that the Observable has no more data and the event queue ends

OnError (): when an exception occurs during event handling, onError () is triggered, and the queue is automatically terminated, and no more events are allowed.

It is precisely because RxJava provides both synchronous and asynchronous ways to handle events, I think the asynchronous way can better reflect the value of RxJava, so it is named asynchronous observer mode.

All right, let's formally introduce those flexible operators of RxJava. Here are just simple introductions and simple examples, which will be introduced in future articles.

Maven introduces io.reactivexrxjava1.2.4 to create Observable

1.create () creates an Observable and defines event trigger rules for it

Observable observable = Observable .create (new Observable.OnSubscribe () {@ Overridepublic void call (Subscriber)

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.