Dubbo 异步调用过程

开篇

• 本篇是基于Dubbo-2.6.7版本的异步调用的分析，在这个过程中会涉及异步的调用过程和响应过程的分析。
• 文章中会有一部分简单的例子，用于讲解异步的是使用方式。

异步调用说明

Dubbo 异步调用过程

• 关注userThread的行为，用户发出调用后，IOThread会在上下文RpcContext中设置Future，对应上图中步骤1.2.3。
• 用户从RpcContext中取得Future，然后wait这个Future其它的事情都由IOThread完成，对应上图中步骤4.5。
• server端响应后会把调用结果设置在RpcContext上下文当中，同时通知UserThread线程。

异步回调使用案例

<dubbo:reference id="fooService" interface="com.alibaba.foo.FooService">
      <dubbo:method name="findFoo" async="true" />
</dubbo:reference>
<dubbo:reference id="barService" interface="com.alibaba.bar.BarService">
      <dubbo:method name="findBar" async="true" />
</dubbo:reference>

// 此方法应该返回Foo，但异步后会立刻返回NULL
fooService.findFoo(fooId);
// 立刻得到当前调用的Future实例，当发生新的调用时这个东西将会被覆盖
Future<Foo> fooFuture = RpcContext.getContext().getFuture();
 
// 调用另一个服务的方法
barService.findBar(barId);
// 立刻得到当前调用的Future
Future<Bar> barFuture = RpcContext.getContext().getFuture();
 
// 此时，两个服务的方法在并发执行
// 等待第一个调用完成，线程会进入Sleep状态，当调用完成后被唤醒。
Foo foo = fooFuture.get();
// 同上
Bar bar = barFuture.get();
// 假如第一个调用需要等待5秒，第二个等待6秒，则整个调用过程完成的时间是6秒。

• 1、异步调用的实现步骤先执行fooService.findFoo()的执行服务调用。
• 2、获取RPC上下文RpcContext.getContext().getFuture()。
• 3、通过future.get()方法获取执行结果，如果当时没有结果当前线程就会被挂起。

Dubbo异步调用栈

• Consumer => InvokerInvocationHandler =>DubboInvoker =>HeaderExchangeClient。

InvokerInvocationHandler

public class InvokerInvocationHandler implements InvocationHandler {

    private final Invoker<?> invoker;

    public InvokerInvocationHandler(Invoker<?> handler) {
        this.invoker = handler;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getDeclaringClass() == Object.class) {
            return method.invoke(invoker, args);
        }
        if ("toString".equals(methodName) && parameterTypes.length == 0) {
            return invoker.toString();
        }
        if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
            return invoker.hashCode();
        }
        if ("equals".equals(methodName) && parameterTypes.length == 1) {
            return invoker.equals(args[0]);
        }
        return invoker.invoke(new RpcInvocation(method, args)).recreate();
    }

}

• InvokerInvocationHandler的invoke()方法创建rpc调用的RpcInvocation对象，这个对象会在单次调用的过程中传递，相当于单次调用的上下文。

RpcInvocation

public class RpcInvocation implements Invocation, Serializable {

    private static final long serialVersionUID = -4355285085441097045L;
    // 方法名字
    private String methodName;
    // 参数类型
    private Class<?>[] parameterTypes;
    // 参数值
    private Object[] arguments;
    // 上下文透传的参数值
    private Map<String, String> attachments;

    private transient Invoker<?> invoker;

    public RpcInvocation() {
    }

    public RpcInvocation(Method method, Object[] arguments) {
        this(method.getName(), method.getParameterTypes(), arguments, null, null);
    }

    public RpcInvocation(Method method, Object[] arguments, Map<String, String> attachment) {
        this(method.getName(), method.getParameterTypes(), arguments, attachment, null);
    }

    public RpcInvocation(String methodName, Class<?>[] parameterTypes, Object[] arguments) {
        this(methodName, parameterTypes, arguments, null, null);
    }

    public RpcInvocation(String methodName, Class<?>[] parameterTypes, Object[] arguments, Map<String, String> attachments) {
        this(methodName, parameterTypes, arguments, attachments, null);
    }

    public RpcInvocation(String methodName, Class<?>[] parameterTypes, Object[] arguments, Map<String, String> attachments, Invoker<?> invoker) {
        this.methodName = methodName;
        this.parameterTypes = parameterTypes == null ? new Class<?>[0] : parameterTypes;
        this.arguments = arguments == null ? new Object[0] : arguments;
        this.attachments = attachments == null ? new HashMap<String, String>() : attachments;
        this.invoker = invoker;
    }
}

RpcInvocation

• RpcInvocation的核心变量包括方法名、参数类型、参数值、附带上下文数据。
• methodName为方法名。
• parameterTypes为参数类型。
• arguments为参数值。
• attachments为附带上下文数据。

异步调用流程

dubbo异步调用.jpg

DubboInvoker

public class DubboInvoker<T> extends AbstractInvoker<T> {
    protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
        inv.setAttachment(Constants.VERSION_KEY, version);

        ExchangeClient currentClient;
        if (clients.length == 1) {
            currentClient = clients[0];
        } else {
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
            // 是否异步
            boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
            // 是否单向
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
            // 超时时间
            int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);

            if (isOneway) {
                // 处理单向发送
                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture(null);
                return new RpcResult();
            } else if (isAsync) {
                // 处理异步发送
                ResponseFuture future = currentClient.request(inv, timeout);
                RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
                return new RpcResult();
            } else {
                // 处理同步发送
                RpcContext.getContext().setFuture(null);
                return (Result) currentClient.request(inv, timeout).get();
            }
        } catch (TimeoutException e) {
            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } catch (RemotingException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }
}

• 异步调用执行ResponseFuture future = currentClient.request(inv, timeout)发送请求。
• 异步调用执行RpcContext.getContext().setFuture(new FutureAdapter<Object>(future))保存future到RpcContext
• 同步调用执行RpcContext.getContext().setFuture(null)设置RpcContext为空。
• 同步调用执行currentClient.request(inv, timeout).get()等待同步消息结果。
• 同步执行和异步执行的差别在于同步发送requst之后执行get()同步等待结果，异步执行发送request之后保存future到RpcContext上下文。

HeaderExchangeChannel

final class HeaderExchangeChannel implements ExchangeChannel {

    private static final Logger logger = LoggerFactory.getLogger(HeaderExchangeChannel.class);

    private static final String CHANNEL_KEY = HeaderExchangeChannel.class.getName() + ".CHANNEL";

    private final Channel channel;

    private volatile boolean closed = false;

    HeaderExchangeChannel(Channel channel) {
        if (channel == null) {
            throw new IllegalArgumentException("channel == null");
        }
        this.channel = channel;
    }

    public ResponseFuture request(Object request, int timeout) throws RemotingException {
        if (closed) {
            throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
        }
        // create request.
        Request req = new Request();
        req.setVersion(Version.getProtocolVersion());
        req.setTwoWay(true);
        req.setData(request);
        // 保存channel、req等信息到DefaultFuture对象当中
        DefaultFuture future = new DefaultFuture(channel, req, timeout);
        try {
            // 调用底层逻辑发送消息
            channel.send(req);
        } catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }
}

• HeaderExchangeChannel的request()方法内部会创建Request对象，核心变量包括version、data的变量。
• channel.send()方法中channel指的是NettyClient对象。
• HeaderExchangeChannel.request()方法返回DefaultFuture对象，用于保存异步至上下文的RpcContext当中
• HeaderExchangeChannel的request()方法内部创建Request对象，创建DefaultFuture对象（包含request对象），调用NettyClient.send()异步发送消息。

Request

public class Request {

    public static final String HEARTBEAT_EVENT = null;

    public static final String READONLY_EVENT = "R";

    private static final AtomicLong INVOKE_ID = new AtomicLong(0);

    private final long mId;

    private String mVersion;

    private boolean mTwoWay = true;

    private boolean mEvent = false;

    private boolean mBroken = false;

    private Object mData;

    public Request() {
        mId = newId();
    }

    public Request(long id) {
        mId = id;
    }

    private static long newId() {
        // getAndIncrement() When it grows to MAX_VALUE, it will grow to MIN_VALUE, and the negative can be used as ID
        return INVOKE_ID.getAndIncrement();
    }

    public long getId() {
        return mId;
    }
}

• Request对象的核心字段INVOKE_ID，全局静态用于记录标识request对象的唯一性。
• Request对象的核心变量如上图所示，其中mData保存RpcInvocation对象。

DefaultFuture

public class DefaultFuture implements ResponseFuture {

    private static final Logger logger = LoggerFactory.getLogger(DefaultFuture.class);
    // 保存等待响应的Channel
    private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<Long, Channel>();
    // 保存等待响应的DefaultFuture
    private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();

    static {
        // 超时检测线程
        Thread th = new Thread(new RemotingInvocationTimeoutScan(), "DubboResponseTimeoutScanTimer");
        th.setDaemon(true);
        th.start();
    }

    // invoke id.
    private final long id;
    private final Channel channel;
    private final Request request;
    private final int timeout;
    // 核心的lock和done字段
    private final Lock lock = new ReentrantLock();
    private final Condition done = lock.newCondition();

    private final long start = System.currentTimeMillis();
    private volatile long sent;
    private volatile Response response;
    private volatile ResponseCallback callback;

    public DefaultFuture(Channel channel, Request request, int timeout) {
        this.channel = channel;
        this.request = request;
        this.id = request.getId();
        this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
        // put into waiting map.
        FUTURES.put(id, this);
        CHANNELS.put(id, channel);
    }

    // 超时检测线程RemotingInvocationTimeoutScan
    private static class RemotingInvocationTimeoutScan implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    // 遍历所有保存的DefaultFuture对象检测超时
                    for (DefaultFuture future : FUTURES.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }

                        // 检测超时的处理逻辑
                        if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                            // 创建Response对象，唯一标识符为Request的唯一标识mId
                            // future.getId() 等价于request.getId()
                            Response timeoutResponse = new Response(future.getId());
                            // set timeout status.
                            timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
                            timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
                            // 调用DefaultFuture.received执行超时响应逻辑
                            DefaultFuture.received(future.getChannel(), timeoutResponse);
                        }
                    }
                    Thread.sleep(30);
                } catch (Throwable e) {
                }
            }
        }
    }

    // 处理数据接受的逻辑或者超时响应的逻辑received => doReceived
    public static void received(Channel channel, Response response) {
        try {
            DefaultFuture future = FUTURES.remove(response.getId());
            if (future != null) {
                future.doReceived(response);
            } else {
            }
        } finally {
            CHANNELS.remove(response.getId());
        }
    }

    // 内部通过done.signal()方式通知等待的线程异步结果。
    private void doReceived(Response res) {
        // 通过lock来实现互斥
        lock.lock();
        try {
            response = res;
            if (done != null) {
                // 通知等待线程
                done.signal();
            }
        } finally {
            lock.unlock();
        }
        // 如果配置回调函数就执行回调函数callback
        if (callback != null) {
            invokeCallback(callback);
        }
    }

    // 以下逻辑是同步等待的逻辑
    public Object get() throws RemotingException {
        return get(timeout);
    }

    @Override
    public Object get(int timeout) throws RemotingException {
        if (timeout <= 0) {
            timeout = Constants.DEFAULT_TIMEOUT;
        }
        if (!isDone()) {
            long start = System.currentTimeMillis();
            lock.lock();
            try {
                while (!isDone()) {
                    done.await(timeout, TimeUnit.MILLISECONDS);
                    if (isDone() || System.currentTimeMillis() - start > timeout) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
            if (!isDone()) {
                throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
            }
        }
        return returnFromResponse();
    }

    private Object returnFromResponse() throws RemotingException {
        Response res = response;
        if (res == null) {
            throw new IllegalStateException("response cannot be null");
        }
        if (res.getStatus() == Response.OK) {
            return res.getResult();
        }
        if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
            throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
        }
        throw new RemotingException(channel, res.getErrorMessage());
    }
}

• DefaultFuture作为异步实现的核心，本质上通过ReentrantLock来实现异步通知。
• Lock lock = new ReentrantLock()，互斥锁用于保证单个DefaultFuture的线程安全。
• Condition done = lock.newCondition()，用于单个DefaultFuture的唤醒通知机制。
• DefaultFuture包含全局唯一的静态线程RemotingInvocationTimeoutScan用于扫描超时的DefaultFuture对象。
• DefaultFuture包含静态变量FUTURES保存所有请求的DefaultFuture对象。
• RemotingInvocationTimeoutScan扫描超时线程后会执行DefaultFuture的received => doReceived流程进行响应。
• 正常响应返回的处理流程会执行DefaultFuture的received => doReceived流程进行响应。
• DefaultFuture的get方法用于执行等待操作，通过done.await()方法实现。

正常返回的处理流程

public class HeaderExchangeHandler implements ChannelHandlerDelegate {

    protected static final Logger logger = LoggerFactory.getLogger(HeaderExchangeHandler.class);

    public static String KEY_READ_TIMESTAMP = HeartbeatHandler.KEY_READ_TIMESTAMP;

    public static String KEY_WRITE_TIMESTAMP = HeartbeatHandler.KEY_WRITE_TIMESTAMP;

    private final ExchangeHandler handler;

    public HeaderExchangeHandler(ExchangeHandler handler) {
        if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        }
        this.handler = handler;
    }

    static void handleResponse(Channel channel, Response response) throws RemotingException {
        if (response != null && !response.isHeartbeat()) {
            DefaultFuture.received(channel, response);
        }
    }

    public void received(Channel channel, Object message) throws RemotingException {
        channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
        ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
        try {
            if (message instanceof Request) {
                // handle request.
                Request request = (Request) message;
                if (request.isEvent()) {
                    handlerEvent(channel, request);
                } else {
                    if (request.isTwoWay()) {
                        Response response = handleRequest(exchangeChannel, request);
                        channel.send(response);
                    } else {
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } else if (message instanceof Response) {
                handleResponse(channel, (Response) message);
            } else if (message instanceof String) {
                if (isClientSide(channel)) {
                    Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
                    logger.error(e.getMessage(), e);
                } else {
                    String echo = handler.telnet(channel, (String) message);
                    if (echo != null && echo.length() > 0) {
                        channel.send(echo);
                    }
                }
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }
}

• 正常响应的时候通过received => handleResponse执行到DefaultFuture的received进行响应。

结论

• Dubbo的异步调用流程的底层核心借助于NettyClient的异步过程。
• Dubbo的异步调用流程的每个请求对象Request都有唯一的标识符（使用递增的数字标识）。
• Dubbo的异步调用流程的核心逻辑通过DefaultFuture来完成，底层逻辑是通过ReentrantLock来实现的。

参考

Dubbo异步调用（七）