How xxl-job Leverages Netty and Dynamic Proxies for High‑Throughput Asynchronous RPC

Communication Layer Overview

xxl-job uses Netty HTTP for communication, although it also supports Mina, Jetty, and Netty TCP; the implementation hard‑codes Netty HTTP.

Overall Process

The complete processing flow is illustrated with an activity diagram.

Key Design Highlights

Dynamic Proxy Pattern – Interfaces ExecutorBiz and AdminBiz encapsulate operations; XxlRpcReferenceBean creates a proxy whose getObject() performs remote communication.

Full Asynchronous Processing – The executor receives messages, deserializes them, and stores task information in a LinkedBlockingQueue. Worker threads consume tasks from the queue, execute them, and place results in a callback queue, reducing Netty worker thread time and increasing throughput.

Asynchronous Wrapper – The code appears synchronous while internally using asynchronous handling.

Scheduler Trigger Code

public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
    ReturnT<String> runResult = null;
    try {
        ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
        // asynchronous processing, final synchronous result
        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("<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 is invoked via the dynamic proxy; the scheduler thread blocks until the executor returns the result.

Dynamic Proxy Implementation

if (CallType.SYNC == callType) {
    XxlRpcFutureResponse futureResponse = new XxlRpcFutureResponse(invokerFactory, xxlRpcRequest, null);
    try {
        client.asyncSend(finalAddress, xxlRpcRequest);
        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 {
        futureResponse.removeInvokerFuture();
    }
}

Future Response Handling

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;
}

Each remote call carries a UUID requestId, which acts as a key to locate the corresponding XxlRpcFutureResponse and wake the blocked thread.