Shulou(Shulou.com)05/31 Report--

This article shows you the RxJava response process analysis of the basic call process is how, the content is concise and easy to understand, absolutely can make your eyes bright, through the detailed introduction of this article, I hope you can get something.

one。 Usage

First, let's look at a simple example:

The running result is:

From the results, it is not difficult to see the overall invocation process:

First, we generate an observer by calling the Observable.create () method, and then we call the map () method to transform the data flow of the original observer to generate a new observer (why the new observer will talk later). Finally, we call the subscribe () method and pass it to our observer, where the observer subscribes to the new observer generated after calling map ().

Throughout the process, we will notice three protagonists: Observable, OnSubscribe, and Subscriber, all of which are operated around them. It is not difficult to see the division of labor among the three roles here:

Observable: the source of the observed, or the observed itself

OnSubscribe: different behaviors used to notify the observer

Subscriber: the observer, which generates concrete processing by implementing the corresponding method.

So next we take these three roles as the center to analyze the specific process.

two。 Analysis 1. Subscription process

First, let's go to Observable.create () and take a look:

Calling the constructor here generates an Observable object and assigns the incoming OnSubscribe to its own member variable onsubscribe, and so on, where did this hook come from? Let's look up:

RxJavaObservableExecutionHook, an abstract Proxy class, does not do any processing on OnSubscribe objects by default, but by inheriting the class and overriding methods such as onCreate (), we can do some extra processing on the timing of these methods, such as typing Log or some data collection.

So far, the original observer has been generated, so let's take a look at the observer side. We know that passing in an observer by calling the observable.subscribe () method constitutes a subscription relationship between the observer and the observed, so how is this internally implemented? Look at the code:

Here we omit part of the extraneous code to look at the main part, subscribe.onStart () default empty implementation we do not care about it, for the incoming subscriber to be packaged as SafeSubscriber, this SafeSubscriber on the original subscriber a series of methods to do a better treatment, including: onError () and onCompleted () only one will be executed; guarantee that once onError () or onCompleted () is executed, will no longer be able to hold onNext () and so on. After encapsulating it as SafeSubscriber, call onSubscribe.call () and pass in the subscriber, thus completing a subscription.

Obviously, as an observer, Subscriber's specific behavior plays a vital role in the chain call after the subscription is completed, so let's take a look at the main parts of its internal composition:

Each Subscriber holds a SubscriptionList, which stores all the observer's subscription events, and the Subscriber also implements the Subscription API. When the Subscriber unsubscribes, it unsubscribes all the Subscription in the list of holding events and no longer accepts any subscription events. At the same time, the total amount of data streams received by the Subscriber can be limited by Producer, which is actually added to the Subscriber.onNext () method, while onComplete () and onError () will not be affected by it. Because it is an underlying abstract class, onNext (), onComplete (), and onError () unification is not handled here.

two。 Transformation process

Before receiving the message from Observable, we may process the data stream, such as map (), flatMap (), deBounce (), buffer () and so on. In this case, we use the map () method, which receives the data sent by the original observer and transmits the result returned by this method as new data, which is equivalent to making an intermediate transformation:

Let's move on to the transformation process:

This is achieved through a lift () method. Looking at other conversion methods, it is found that lift () is also used internally. It seems that this lift () is the key, but there is no hurry. Let's take a look at what this OperationMap is:

OperationMap implements the call () method of the Operator interface, which accepts incoming observers and constructs a new observer as a parameter. It is not difficult to find o.onNext (transformer.call (t)); this sentence plays a vital role, where the interface transformer converts generic T to generic R:

After that, the converted data is returned to the original observer's onNext () method, and the observation data flow is converted, but you should also notice that the new observer we used to do the transformation does not implement the subscription operation of the observed. Where is this subscription operation implemented? The answer is the next lift ():

Here we generate a new Observable object. In the call () method of the onSubscribe member of this new object, we get the previously generated non-subscription observer st through operator.call (), and then pass it as an argument to the initial onSubscribe.call (), which completes the process of intermediate subscription.

Now let's sort out the whole process:

One map () transformation

Generate a new Subscriber based on the Operator instance

Generate a new Observable through lift ()

The original Subscriber subscribes to the new Observavble

OnSubscribe Notification in New Observable New Subscriber subscription to the original Observable

The new Subscriber passes the message to the original Subscriber.

To make it easier to understand, let's borrow a picture of the throwing line:

This is the process of a map () transformation, and in fact the same is true of map () several times: after the outermost target Subscriber subscribes, onSubscribe.onNext () will be called layer by layer, until the initial Observable is subscribed by the lowest Subscriber, and the message will be sent to the target Subscriber through layer by layer changes in Operator. Show the picture of the throwing line again:

As for many other changes, the implementation process is also very similar, with the help of different implementations of Operator to achieve the purpose of transforming the data stream. For example, flatMap (), which needs to do lift () twice, the second time is OperationMerge. After subscribing each Observable data stream into InnerSubscriber, it gets R in the onNext () of InnerSubscriber, and then sends them all out (emit) through the incoming parent (that is, the original MergeSubscriber), and receives them uniformly by the outermost Subscriber we have passed in. Thus the conversion of T = > Observable = > R is completed.

In addition, there are a variety of operators, and if they don't meet your needs, you can also customize new operators by implementing the Operator interface. Flexible use of them can often achieve twice the result with half the effort, such as effectively avoiding the need for backpressure by using operators such as sample (), debounce (), and so on.

The above is the RxJava response process analysis of the basic invocation process. Have you learned any knowledge or skills? If you want to learn more skills or enrich your knowledge reserve, you are welcome to follow the industry information channel.

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.