1.1 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, get the response data. Inverse geographic interface: obtain the province, region and county where the longitude and latitude are located and the response code by latitude and longitude:

curl-i"http://xxx? Latitude = 31.08966221524924 & channel = amap7a & near = false&longitude = 105.13990312814713"Copy the code
{"adcode":"510722"}
Copy the code

Server-side execution, the simplest synchronous invocation:Before the server response, IO blocks in: java.net.SocketInputStream#socketRead0 native methods:Jstack logs show that the Thread is always in the runable state:

"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)
                .......
Copy the code

Example thread model:The biggest problem with synchronization is that thread resources are not fully utilized during I/O waiting, which limits service throughput in many I/O scenarios.

2.1 JDK NIO & Future

In JDK 1.5, JUC provides a Future abstraction:

, of course, not all of the Future is always the same, such as io.net. Ty util. Concurrent. AbstractFuture thread is through polling.

The advantage of this is that the main thread does not have to wait for an I/O response. It can do something else, such as send another I/O request, and wait until it returns:

"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  <0x000000076ee2d768> (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 <0x000000076eb00ef0> (a io.netty.channel.nio.SelectedSelectionKeySet)
- locked <0x000000076eb00f10> (a java.util.Collections$UnmodifiableSet)
- locked <0x000000076eb00ea0> (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)
Copy the code

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

3.1 Using the Callback Mode

In section 2, the main thread is still required to wait for the result, so can the main thread finish sending the request, no longer care about this logic, to execute other logic? Then you can use the Callback mechanism.

This way, the main thread no longer needs to care about the business logic after the IO is initiated, and after the request is sent, it can either do something else entirely or return to the thread pool for scheduling. If it is HttpServer, then asynchronous servlets with Servlet 3.1 are required.

Asynchronous servlets resources www.cnblogs.com/davenkin/p/…

4.1 the Callback hell

Callback hell, when the Callback thread needs to perform the next IO call.

A typical application scenario is to obtain the adcode (inverse geographic interface) of the administrative region based on the longitude and latitude, and then obtain the local weather information based on the adcode (weather interface).

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

5.1 JDK 1.8 CompletableFuture

So is there a way around Callback Hell? Of course, the CompletableFuture is provided in JDK 1.8, so let’s take a look at how it solves this problem.

Encapsulate the inverse geography Callback logic into a separate CompletableFuture. When the asynchronous thread calls back, future.plete (T) is called to encapsulate the result.

Encapsulate the Call logic for weather execution into a separate CompletableFuture, and when it’s done, the logic is the same as above.

Compose connection, whenComplete output:

Each IO operation can be encapsulated as a separate CompletableFuture to avoid callback hell. CompletableFuture has only two properties:

  • Result: Future execution result (Either the result or boxed AltResult).
  • Stack: The stack of actions that defines the Top of Treiber stack of dependent actions for the Future.

How is the weatherFuture method called?

Through the stack can be found, in reverseCodeFuture.com plete (result), and will also get adcode as parameters to perform the following logic.In this way, the perfect solution to the callback hell problem, in the Lord’s logic, seems to be synchronous coding.

The Future Vert. 6.1 x

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

The logic of core execution is similar:

That’s certainly not all of Vertx, but that’s a digression of course.

7.1 Reactive Streams

Asynchronous programming is good for throughput and resources, but if there is a unified abstraction to address these issues, Reactive Streams is the answer.

Core abstraction: Publisher Subscriber Processor Subscription, only four interfaces, no implementation classes.

In JDK 9, has already been packaged as a specification to Java. The util. Concurrent. The Flow:

Resources www.baeldung.com/java-9-reac… Ypk1226.com/2019/07/01/… www.reactivemanifesto.org/ projectreactor.io/learn

A simple example:

8.1 Reactor & Spring 5 & Spring WebFlux

Flux & Mono

Resources projectreactor. IO/docs/core / 3… Speakerdeck.com/simonbasle/…