淺談異步併發請求和請求合併

時間 2019-11-13

標籤淺談異步併發求和請求合併简体版

原文原文鏈接

概述

在作業務系統需求開發中，常常須要從其餘服務獲取數據，拼接數據，而後返回數據給前端使用；常見的服務調用就是經過http接口調用，而對於http，一般一個請求會分配一個線程執行，在同步調用接口的狀況下，整個線程是一直被佔用或者阻塞的；若是有大量的這種請求，整個系統的吞吐量就比較低，而在依賴的服務響應時間比較低的狀況下，咱們但願先讓出cpu，讓其餘請求先執行，等依賴的服務請求返回結果時再繼續往下執行，這時咱們會考慮將請求異步化，或者將相同的請求合併，從而達到提升系統執行效率和吞吐量的目的。前端

異步併發請求

目前常見的幾種調用方式是同步調用，線程池+future，異步回調completableFuture；協程也是異步調用的解決方式，但java目前不支持協程；對於future方式，只能用get或者while(!isDone)輪詢這種阻塞的方式直到線程執行完成，這也不是咱們但願的異步執行方式，jdk8提供的completableFuture其實也不是異步的方式，只是對依賴多服務的Callback調用結果處理作結果編排，來彌補Callback的不足，從而實現異步鏈式調用的目的，這也是比較推薦的方式。java

同步調用

RpcService rpcService = new RpcService();
	HttpService httpService = new HttpService();
	// 假設rpc1耗時10ms
	Map<String, String> rpcResult1=rpcService.getRpcResult();
	// 假設rpc2耗時20ms
	Integer rpcResult2 = httpService.getHttpResult();
	
	// 則線程總耗時：30ms

future

ExecutorService executor = Executors.newFixedThreadPool(2);
	RpcService rpcService = new RpcService();
	HttpService httpService = new HttpService();
	future1 = executor.submit(() -> rpcService.getRpcResult());
	future2 = executor.submit(() -> httpService.getHttpResult());
	//rpc1耗時10ms
	Map<String, String> rpcResult1 = future1.get(300, TimeUnit.MILLISECONDS);
	//rpc2耗時20ms
	Integer rpcResult2 = future2.get(300, TimeUnit.MILLISECONDS);

	//則線程總耗時20ms

CompletableFuture

/** 
	* 場景1：兩個接口併發異步調用，返回CompletableFuture,不阻塞主線程 
	* 兩個服務也是異步非阻塞調用
	**/
	CompletableFuture future1 = service.getHttpData("http://www.vip.com/showGoods/50");
	CompletableFuture future2 = service.getHttpData("http://www.vip.com/showGoods/50");
	CompletableFuture future3 = future1.thenCombine(future2, (f1, f2) -> {
	    //處理業務....

	    return f1 + "," + f2;
	}).exceptionally(e -> {

	    return "";
	});

CompletableFuture使用ForkJoinPool執行線程，在ForkJoinPool類註冊ForkJoinWorkerThread線程時能夠看到，ForkJoinPool裏面的線程都是daemon線程（垃圾回收線程就是一個典型的deamon線程），segmentfault

/**
     * Callback from ForkJoinWorkerThread constructor to establish and
     * record its WorkQueue.
     *
     * @param wt the worker thread
     * @return the worker's queue
     */
    final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
        UncaughtExceptionHandler handler;
		
	#註冊守護線程
        wt.setDaemon(true);          // configure thread
		
        if ((handler = ueh) != null)
            wt.setUncaughtExceptionHandler(handler);
        WorkQueue w = new WorkQueue(this, wt);
        int i = 0;                                    // assign a pool index
        int mode = config & MODE_MASK;
        int rs = lockRunState();
        try {
            WorkQueue[] ws; int n;                    // skip if no array
            if ((ws = workQueues) != null && (n = ws.length) > 0) {
                int s = indexSeed += SEED_INCREMENT;  // unlikely to collide
                int m = n - 1;
                i = ((s << 1) | 1) & m;               // odd-numbered indices
                if (ws[i] != null) {                  // collision
                    int probes = 0;                   // step by approx half n
                    int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
                    while (ws[i = (i + step) & m] != null) {
                        if (++probes >= n) {
                            workQueues = ws = Arrays.copyOf(ws, n <<= 1);
                            m = n - 1;
                            probes = 0;
                        }
                    }
                }
                w.hint = s;                           // use as random seed
                w.config = i | mode;
                w.scanState = i;                      // publication fence
                ws[i] = w;
            }
        } finally {
            unlockRunState(rs, rs & ~RSLOCK);
        }
        wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));
        return w;
    }

當主線程執行完後，jvm就會退出，因此須要考慮主線程執行完成時間和fork出去的線程執行時間，也須要考慮線程池的大小，默認爲當前cpu的核數-1，能夠參考下其餘系統的故障記錄：CompletableFuture線程池問題。緩存