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

This article is about how to achieve asynchronous non-blocking programming in Java. The editor thinks it is very practical, so I share it with you. I hope you can get something after reading this article.

Start with a synchronous Http call

A very simple business logic, other back-end services provide an interface, we need to call through the interface to get the response data.

Inverse geographic interface: obtain the province, city, autonomous region and county where the longitude and latitude are located and the response code through longitude and latitude:

Curl-i "http://xxx?latitude=31.08966221524924&channel=amap7a&near=false&longitude=105.13990312814713"{"adcode":"510722"}

Server execution, the simplest synchronous invocation method:

Before the server responds, the IO blocks at:

On the native method of java.net.SocketInputStream#socketRead0:

Through the jstack log, you can find that the Thread will always be in the state of runable:

"main" # 1 prio=5 os_prio=31 tid=0x00007fed0c810000 nid=0x1003 runnable [0x000070000ce14000] java.lang.Thread.State: RUNNABLE at java.net.SocketInputStream.socketRead0 (Native Method) at java.net.SocketInputStream.socketRead (SocketInputStream.java:116) at java.net.SocketInputStream.read (SocketInputStream.java:171) at java.net.SocketInputStream.read (SocketInputStream.java:141) at org.apache.http.impl.conn.LoggingInputStream.read (LoggingInputStream.java: 84) at org.apache.http.impl.io.SessionInputBufferImpl.streamRead (SessionInputBufferImpl.java:137) at org.apache.http.impl.io.SessionInputBufferImpl.fillBuffer (SessionInputBufferImpl.java:153) at org.apache.http.impl.io.SessionInputBufferImpl.readLine (SessionInputBufferImpl.java:282) at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead (DefaultHttpResponseParser.java:138) at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead (DefaultHttpResponseParser.java:56) at org.apache.http.impl.io.AbstractMessageParser.parse (AbstractMessageParser.java:259) at org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader (DefaultBHttpClientConnection.java:163) at org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader (CPoolProxy.java:165) at org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse (HttpRequestExecutor.java:273) at org.apache.http.protocol.HttpRequestExecutor.execute ( HttpRequestExecutor.java:125) at org.apache.http.impl.execchain.MainClientExec.execute (MainClientExec.java:272) at org.apache.http.impl.execchain.ProtocolExec.execute (ProtocolExec.java:185) at org.apache.http.impl.execchain.RetryExec.execute (RetryExec.java:89) at org.apache.http.impl.execchain.RedirectExec.execute (RedirectExec.java:110) at org.apache.http.impl.client .InternalHttpClient.doExecute (InternalHttpClient.java:185) at org.apache.http.impl.client.CloseableHttpClient.execute (CloseableHttpClient.java:83) at org.apache.http.impl.client.CloseableHttpClient.execute (CloseableHttpClient.java:108) at com.amap.aos.async.AsyncIO.blockingIO (AsyncIO.java:207).

Example of a threading model:

The biggest problem of synchronization is that thread resources are not fully utilized in the process of IO waiting, which will limit the business throughput of a large number of IO scenarios.

Two JDK NIO & Future

In JDK 1.5, JUC provides the Future abstraction:

Of course, not all Future are implemented in this way, such as

Io.netty.util.concurrent.AbstractFuture is polled by threads.

The advantage of this is that instead of waiting for an IO response, the main thread can do something else, such as sending another IO request, and waiting for it to return together:

"main" # 1 prio=5 os_prio=31 tid=0x00007fd7a500b000 nid=0xe03 waiting on condition [0x000070000a95d000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park (Native Method)-parking to wait for (a java.util.concurrent.CountDownLatch$Sync) at java.util.concurrent.locks.LockSupport.park (LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt (AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks .AbstractQueuedSynchronizer.doAcquireSharedInterruptibly (AbstractQueuedSynchronizer.java:997) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly (AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.CountDownLatch.await (CountDownLatch.java:231) at org.asynchttpclient.netty.NettyResponseFuture.get (NettyResponseFuture.java:162) at com.amap.aos.async.AsyncIO.futureBlockingGet (AsyncIO.java:201). "AsyncHttpClient-2-1" # 11 prio=5 os_ Prio=31 tid=0x00007fd7a7247800 nid=0x340b runnable [0x000070000ba94000] java.lang.Thread.State: RUNNABLE at sun.nio.ch.KQueueArrayWrapper.kevent0 (Native Method) at sun.nio.ch.KQueueArrayWrapper.poll (KQueueArrayWrapper.java:198) at sun.nio.ch.KQueueSelectorImpl.doSelect (KQueueSelectorImpl.java:117) at sun.nio.ch.SelectorImpl.lockAndDoSelect (SelectorImpl.java:86)-locked (an io.netty.channel.nio.SelectedSelectionKeySet)-locked (a java.util .Collections $UnmodifiableSet)-locked (a sun.nio.ch.KQueueSelectorImpl) at sun.nio.ch.SelectorImpl.select (SelectorImpl.java:97) at io.netty.channel.nio.NioEventLoop.select (NioEventLoop.java:693) at io.netty.channel.nio.NioEventLoop.run (NioEventLoop.java:353) at io.netty.util.concurrent.SingleThreadEventExecutor$2.run (SingleThreadEventExecutor.java:140) at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator .run (DefaultThreadFactory.java:144) at java.lang.Thread.run (Thread.java:748)

The main thread still needs to wait while waiting for the result to return, which does not fundamentally solve this problem.

Third, use Callback callback

In the second section, you still need the main thread to wait and get the results, so can you no longer care about this logic and execute other logic after the main thread has finished sending the request? Then you can use the Callback mechanism.

In this way, the main thread no longer needs to care about the business logic after initiating the IO, and after sending the request, it can completely do something else, or go back to the thread pool for scheduling. If it is HttpServer, then asynchronous Servlet combined with Servlet 3.1 is required.

Asynchronous Servelt reference material

Https://www.cnblogs.com/davenkin/p/async-servlet.html

Using Callback method, from the thread model, it is found that thread resources have been fully utilized, and there is no thread blocking in the whole process.

Four Callback hell

Callback hell, which enters callback hell mode when the Callback thread needs to make the next IO call.

Typical application scenarios, for example, obtain administrative region adcode (inverse geographic interface) through longitude and latitude, and then obtain local weather information (weather interface) according to the obtained adcode.

In the synchronous programming model, this kind of problem is rarely involved.

Core defects of the Callback approach

Five JDK 1.8 CompletableFuture

So is there any way to solve the problem of Callback Hell? Of course, CompletableFuture is provided in JDK 1.8, so let's see how it solves this problem.

The Callback logic of reverse geography is encapsulated into a separate CompletableFuture. When the asynchronous thread is called back, future.complete (T) is called to encapsulate the result.

The Call logic executed by the weather is also encapsulated into a separate CompletableFuture, and after completion, the logic is the same as above.

Compose connection, whenComplete output:

Each IO operation can be encapsulated as a separate CompletableFuture, thus avoiding callback hell.

CompletableFuture, there are only two properties:

The execution result of result:Future (Either the result or boxed AltResult).

Stack: operation stack, which defines the behavior (Top of Treiber stack of dependent actions) of the next operation of this Future.

How is the weatherFuture method called?

As can be found through the stack, it is found in the

ReverseCodeFuture.complete (result), and also take the obtained adcode as a parameter to execute the following logic.

In this way, the callback hell problem is solved perfectly, and in the main logic, it seems to be coding synchronously.

Six Vert.x Future

Vert.x Future, which is widely used in Info-Service, is a similar solution, but is designed to use the concept of Handler.

The logic of core execution is similar:

This is certainly not the whole story of Vertx, but of course it is beside the point.

Seven Reactive Streams

Asynchronous programming is good for throughput and resources, but there is no unified abstraction to solve such problems, the answer is Reactive Streams.

Core abstraction: Publisher Subscriber Processor Subscription, in the whole package, there are only these four interfaces, no implementation classes.

In JDK 9, it has been encapsulated into java.util.concurrent.Flow as a specification:

references

Https://www.baeldung.com/java-9-reactive-streams

Http://ypk1226.com/2019/07/01/reactive/reactive-streams/

Https://www.reactivemanifesto.org/

Https://projectreactor.io/learn

A simple example:

Eight Reactor & Spring 5 & Spring WebFlux

Flux & Mono

The above is how to achieve asynchronous non-blocking programming in Java. The editor believes that there are some knowledge points that we may see or use in our daily work. I hope you can learn more from this article. For more details, please 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.