Java精選面試題（微信小程序）：5000+道面試題和選擇題，包含Java基礎(chǔ)、并發(fā)、JVM、線程、MQ系列、Redis、Spring系列、Elasticsearch、Docker、K8s、Flink、Spark、架構(gòu)設(shè)計(jì)、大廠真題等，在線隨時(shí)刷題！

1、通信底層介紹

xxl-job 使用 netty http 的方式進(jìn)行通信，雖然也支持 Mina，jetty，netty tcp 等方式，但是代碼里面固定寫(xiě)死的是 netty http。

2、通信整體流程

我以調(diào)度器通知執(zhí)行器執(zhí)行任務(wù)為例，繪制的活動(dòng)圖：

3、驚艷的設(shè)計(jì)

看完了整個(gè)處理流程代碼，設(shè)計(jì)上可以說(shuō)獨(dú)具匠心，將 netty，多線程的知識(shí)運(yùn)用得行云流水。

我現(xiàn)在就將這些設(shè)計(jì)上出彩的點(diǎn)總結(jié)如下：

使用動(dòng)態(tài)代理模式，隱藏通信細(xì)節(jié)

xxl-job 定義了兩個(gè)接口 ExecutorBiz，AdminBiz，ExecutorBiz 接口中封裝了向心跳，暫停，觸發(fā)執(zhí)行等操作，AdminBiz 封裝了回調(diào)，注冊(cè)，取消注冊(cè)操作，接口的實(shí)現(xiàn)類中，并沒(méi)有通信相關(guān)的處理。

XxlRpcReferenceBean 類的 getObject() 方法會(huì)生成一個(gè)代理類，這個(gè)代理類會(huì)進(jìn)行遠(yuǎn)程通信。

全異步處理

執(zhí)行器收到消息進(jìn)行反序列化，并沒(méi)有同步執(zhí)行任務(wù)代碼，而是將任務(wù)信息存儲(chǔ)在 LinkedBlockingQueue 中，異步線程從這個(gè)隊(duì)列中獲取任務(wù)信息，然后執(zhí)行。

而任務(wù)的處理結(jié)果，也不是說(shuō)處理完之后，同步返回的，也是放到回調(diào)線程的阻塞隊(duì)列中，異步的將處理結(jié)果返回回去。

這樣處理的好處就是減少了 netty 工作線程的處理時(shí)間，提升了吞吐量。

對(duì)異步處理的包裝

對(duì)異步處理進(jìn)行了包裝，代碼看起來(lái)是同步調(diào)用的。

我們看下調(diào)度器，XxlJobTrigger 類觸發(fā)任務(wù)執(zhí)行的代碼：

public static ReturnT 

   
runExecutor(TriggerParam triggerParam, String address){
    ReturnT 

 runResult =  
null;
    try {
        ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
        //這里面做了很多異步處理，最終同步得到處理結(jié)果
        runResult = executorBiz.run(triggerParam);
    } catch (Exception e) {
        logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e);
        runResult = new ReturnT 

 (ReturnT.FAIL_CODE, ThrowableUtil.toString(e)); 

    }

StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "：");
    runResultSB.append("
address：").append(address);
    runResultSB.append("
code：").append(runResult.getCode());
    runResultSB.append("
msg：").append(runResult.getMsg());

runResult.setMsg(runResultSB.toString());
    return runResult;
}

ExecutorBiz.run 方法我們說(shuō)過(guò)了，是走的動(dòng)態(tài)代理，和執(zhí)行器進(jìn)行通信，執(zhí)行器執(zhí)行結(jié)果也是異步處理完，才返回的，而這里看到的 run 方法是同步等待處理結(jié)果返回。

我們看下xxl-job是如何同步獲取處理結(jié)果的：調(diào)度器向執(zhí)行器發(fā)出消息后，該線程阻塞。等到執(zhí)行器處理完畢后，將處理結(jié)果返回，喚醒被阻塞的線程，調(diào)用處拿到返回值。

動(dòng)態(tài)代理代碼如下：

//代理類中的觸發(fā)調(diào)用
if (CallType.SYNC == callType) {
   // future-response set
   XxlRpcFutureResponse futureResponse = new XxlRpcFutureResponse(invokerFactory, xxlRpcRequest, null);
   try {
      // do invoke
      client.asyncSend(finalAddress, xxlRpcRequest);

// future get
      XxlRpcResponse xxlRpcResponse = futureResponse.get(timeout, TimeUnit.MILLISECONDS);
      if (xxlRpcResponse.getErrorMsg() != null) {
         thrownew XxlRpcException(xxlRpcResponse.getErrorMsg());
      }
      return xxlRpcResponse.getResult();
   } catch (Exception e) {
      logger.info(">>>>>>>>>>> xxl-rpc, invoke error, address:{}, XxlRpcRequest{}", finalAddress, xxlRpcRequest);

throw (e instanceof XxlRpcException)?e:new XxlRpcException(e);
   } finally{
      // future-response remove
      futureResponse.removeInvokerFuture();
   }
} 

XxlRpcFutureResponse 類中實(shí)現(xiàn)了線程的等待，和線程喚醒的處理：

//返回結(jié)果，喚醒線程
public void setResponse(XxlRpcResponse response) {
   this.response = response;
   synchronized (lock) {
      done = true;
      lock.notifyAll();
   }
}

@Override
    public XxlRpcResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (!done) {
            synchronized (lock) {
                try {
                    if (timeout < 0) {
            //線程阻塞
                        lock.wait();
                    } else {
                        long timeoutMillis = (TimeUnit.MILLISECONDS==unit)?timeout:TimeUnit.MILLISECONDS.convert(timeout , unit);
                        lock.wait(timeoutMillis);
                    }
                } catch (InterruptedException e) {
                    throw e;
                }
            }
        }

if (!done) {
            thrownew XxlRpcException("xxl-rpc, request timeout at:"+ System.currentTimeMillis() +", request:" + request.toString());
        }
        return response;
    }

有的同學(xué)可能會(huì)問(wèn)了，調(diào)度器接收到返回結(jié)果，怎么確定喚醒哪個(gè)線程呢？

每一次遠(yuǎn)程調(diào)用，都會(huì)生成 uuid 的請(qǐng)求 id，這個(gè) id 是在整個(gè)調(diào)用過(guò)程中一直傳遞的，就像一把鑰匙，在你回家的的時(shí)候，拿著它就帶開(kāi)門。

這里拿著請(qǐng)求 id 這把鑰匙，就能找到對(duì)應(yīng)的 XxlRpcFutureResponse，然后調(diào)用 setResponse 方法，設(shè)置返回值，喚醒線程。

public void notifyInvokerFuture(String requestId, final XxlRpcResponse xxlRpcResponse){

// 通過(guò)requestId找到XxlRpcFutureResponse，
    final XxlRpcFutureResponse futureResponse = futureResponsePool.get(requestId);
    if (futureResponse == null) {
        return;
    }
    if (futureResponse.getInvokeCallback()!=null) {

// callback type
        try {
            executeResponseCallback(new Runnable() {
                @Override
                public void run() {
                    if (xxlRpcResponse.getErrorMsg() != null) {
                        futureResponse.getInvokeCallback().onFailure(new XxlRpcException(xxlRpcResponse.getErrorMsg()));
                    } else {
                        futureResponse.getInvokeCallback().onSuccess(xxlRpcResponse.getResult());
                    }
                }
            });
        }catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    } else {
        // 里面調(diào)用lock的notify方法
        futureResponse.setResponse(xxlRpcResponse);
    }

// do remove
    futureResponsePool.remove(requestId);

}

來(lái)源：https://blog.csdn.net/weixin_45334346?type=blog

公眾號(hào)“Java精選”所發(fā)表內(nèi)容注明來(lái)源的，版權(quán)歸原出處所有（無(wú)法查證版權(quán)的或者未注明出處的均來(lái)自網(wǎng)絡(luò)，系轉(zhuǎn)載，轉(zhuǎn)載的目的在于傳遞更多信息，版權(quán)屬于原作者。如有侵權(quán)，請(qǐng)聯(lián)系，筆者會(huì)第一時(shí)間刪除處理！

最近有很多人問(wèn)，有沒(méi)有讀者或者摸魚(yú)交流群！加入方式很簡(jiǎn)單，公眾號(hào)Java精選，回復(fù)“加群”，即可入群！

文章有幫助的話，點(diǎn)在看，轉(zhuǎn)發(fā)吧！