通信底层介绍
xxl-job 应用 netty http 的形式进行通信,尽管也反对 Mina,jetty,netty tcp 等形式,然而代码外面固定写死的是 netty http。
通信整体流程
我以调度器告诉执行器执行工作为例,绘制的流动图:
流动图
惊艳的设计
看完了整个解决流程代码,设计上能够说步人后尘,将 netty,多线程的常识使用得行云流水。
我当初就将这些设计上出彩的点总结如下:
| 应用动静代理模式,暗藏通信细节
xxl-job 定义了两个接口 ExecutorBiz,AdminBiz,ExecutorBiz 接口中封装了向心跳,暂停,触发执行等操作,AdminBiz 封装了回调,注册,勾销注册操作,接口的实现类中,并没有通信相干的解决。
XxlRpcReferenceBean 类的 getObject() 办法会生成一个代理类,这个代理类会进行近程通信。
| 全异步解决
执行器收到音讯进行反序列化,并没有同步执行工作代码,而是将工作信息存储在 LinkedBlockingQueue 中,异步线程从这个队列中获取工作信息,而后执行。
而工作的处理结果,也不是说解决完之后,同步返回的,也是放到回调线程的阻塞队列中,异步的将处理结果返回回去。
这样解决的益处就是缩小了 netty 工作线程的解决工夫,晋升了吞吐量。
| 对异步解决的包装
对异步解决进行了包装,代码看起来是同步调用的。
咱们看下调度器,XxlJobTrigger 类触发工作执行的代码:
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){ ReturnT<String> runResult = null; try { ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address); //这外面做了很多异步解决,最终同步失去处理结果 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<String>(ReturnT.FAIL_CODE, ThrowableUtil.toString(e)); } StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + ":"); runResultSB.append("<br>address:").append(address); runResultSB.append("<br>code:").append(runResult.getCode()); runResultSB.append("<br>msg:").append(runResult.getMsg()); runResult.setMsg(runResultSB.toString()); return runResult;}
ExecutorBiz.run 办法咱们说过了,是走的动静代理,和执行器进行通信,执行器执行后果也是异步解决完,才返回的,而这里看到的 run 办法是同步期待处理结果返回。
咱们看下xxl-job是如何同步获取处理结果的:调度器向执行器收回音讯后,该线程阻塞。等到执行器处理完毕后,将处理结果返回,唤醒被阻塞的线程,调用处拿到返回值。
动静代理代码如下:
//代理类中的触发调用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) { throw new 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 类中实现了线程的期待,和线程唤醒的解决:
//返回后果,唤醒线程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) { throw new XxlRpcException("xxl-rpc, request timeout at:"+ System.currentTimeMillis() +", request:" + request.toString()); } return response; }
有的同学可能会问了,调度器接管到返回后果,怎么确定唤醒哪个线程呢?
每一次近程调用,都会生成 uuid 的申请 id,这个 id 是在整个调用过程中始终传递的,就像一把钥匙,在你回家的的时候,拿着它就带开门。
这里拿着申请 id 这把钥匙,就能找到对应的 XxlRpcFutureResponse,而后调用 setResponse 办法,设置返回值,唤醒线程。
public void notifyInvokerFuture(String requestId, final XxlRpcResponse xxlRpcResponse){ // 通过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 { // 外面调用lock的notify办法 futureResponse.setResponse(xxlRpcResponse); } // do remove futureResponsePool.remove(requestId);}