聊聊java高并发系统之异步非阻塞

àì夳堔傛蜴生んèń 2022-09-26 02:13 141阅读 0赞

**聊聊java高并发系统之异步非阻塞**

**几种调用方式**

**同步阻塞调用**

即串行调用，响应时间为所有服务的响应时间总和；

**半异步(异步Future)**

线程池，异步Future，使用场景：并发请求多服务，总耗时为最长响应时间；提升总响应时间，但是阻塞主请求线程，高并发时依然会造成线程数过多，CPU上下文切换；

**全异步(Callback)**

Callback方式调用，使用场景：不考虑回调时间且只能对结果做简单处理，如果依赖服务是两个或两个以上服务，则不能合并两个服务的处理结果；不阻塞主请求线程，但使用场景有限。

**异步回调链式编排**

异步回调链式编排（JDK8 CompletableFuture），使用场景：其实不是异步调用方式，只是对依赖多服务的Callback调用结果处理做结果编排，来弥补Callback的不足，从而实现全异步链式调用。

接下来看看如何设计利用全异步Callback调用和异步回调链式编排处理结果来实现全异步系统设计。

同步阻塞调用：

public class Test {
       public static void main(String[] args) throws Exception {
           RpcService rpcService = new RpcService();
           HttpService httpService = new HttpService();
           //耗时10ms
           Map result1 = rpcService.getRpcResult();
           //耗时20ms
           Integer result2 = httpService.getHttpResult();
           //总耗时30ms
        }
       static class RpcService {
           Map getRpcResult() throws Exception {
               //调用远程方法（远程方法耗时约10ms，可以使用Thread.sleep模拟）
           }
        }
       static class HttpService {
           Integer getHttpResult() throws Exception {
               //调用远程方法（远程方法耗时约20ms，可以使用Thread.sleep模拟）
               Thread.sleep(20);
               return 0;
           }
        }
    }

**  半异步（异步Future）:**

public class Test {
       final static ExecutorService executor = Executors.newFixedThreadPool(2);
       public static void main(String[] args) {
           RpcService rpcService = new RpcService();
           HttpService httpService = new HttpService();
           Future> future1 = null;
           Future future2 = null;
           try {
               future1 = executor.submit(() -> rpcService.getRpcResult());
               future2 = executor.submit(() -> httpService.getHttpResult());
               //耗时10ms
               Map result1 = future1.get(300, TimeUnit.MILLISECONDS);
               //耗时20ms
               Integer result2 = future2.get(300, TimeUnit.MILLISECONDS);
               //总耗时20ms
           } catch (Exception e) {
               if (future1 != null) {
                    future1.cancel(true);
               }
               if (future2 != null) {
                    future2.cancel(true);
               }
               throw new RuntimeException(e);
           }
        }
       static class RpcService {
           Map getRpcResult() throws Exception {
               //调用远程方法（远程方法耗时约10ms，可以使用Thread.sleep模拟）
           }
        }
       static class HttpService {
           Integer getHttpResult() throws Exception {
               //调用远程方法（远程方法耗时约20ms，可以使用Thread.sleep模拟）
           }
        }
    }<strong>
    
    </strong>

**全异步（callback）:**

public class AsyncTest {
    public staticHttpAsyncClient httpAsyncClient;
       public static CompletableFuture getHttpData(String url) {
           CompletableFuture asyncFuture = new CompletableFuture();
           HttpPost post = new HttpPost(url);
           HttpAsyncRequestProducer producer = HttpAsyncMethods.create(post);
           AsyncCharConsumer consumer = newAsyncCharConsumer() {
                HttpResponse response;
               protected HttpResponse buildResult(final HttpContext context) {
                    return response;
               }
    …...
           };
           FutureCallback callback = new FutureCallback() {
               public void completed(HttpResponse response) {
                   asyncFuture.complete(EntityUtils.toString(response.getEntity()));
               }
    …...
           };
           httpAsyncClient.execute(producer, consumer, callback);
           return asyncFuture;
        }
     
       public static void main(String[] args) throws Exception {
           AsyncTest.getHttpData("http://www.jd.com");
           Thread.sleep(1000000);
        }
    }

**异步回调链式编排:**

CompletableFuture提供了50多个API，可以满足所需的各种场景的异步处理的编排，下面举一个场景：

三个服务并发异步调用，返回CompletableFuture，不阻塞主线程；

![Center][]

public static void test1() throws Exception {
           HelloClientDemoTest service = new HelloClientDemoTest();
           /**
            * 场景1 两个以上服务并发异步调用，返回CompletableFuture,不阻塞主线程
            * 并且两个服务也是异步非阻塞调用
            */
           CompletableFuture future1 = service.getHttpData("http://www.jd.com");
           CompletableFuture future2 = service.getHttpData("http://www.jd.com");
           CompletableFuture future3 =service.getHttpData("http://www.jd.com");
           List futureList = Lists.newArrayList(future1,future2, future3);
           CompletableFuture allDoneFuture =CompletableFuture.allOf(futureList.toArray(newCompletableFuture[futureList.size()]));
           CompletableFuture future4 =allDoneFuture.thenApply(v -> {
                List result =futureList.stream().map(CompletableFuture::join)
    
                       .collect(Collectors.toList());
                //注意顺序
                String result1 = (String)result.get(0);
                String result2 = (String)result.get(1);
                String result3 = (String)result.get(2);
                //处理业务....
                return result1 + result2 + result3;
            }).exceptionally(e -> {
                //e.printStackTrace();
                return "";
            });
           //返回
        }

[Center]: /images/20220717/c1911eb00b494138ab386b7b3dc00e6e.png