dubbo学习远程调用原理

Posted 2021-04-05 码云大作战

一、Dubbo远程调用流程

    Cluster.join - 服务列表聚合代码：

org.apache.dubbo.rpc.cluster.Cluster#join

    这里由于没配置路由规则，因此默认会走FailoverCluster实现。

public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
    return new FailoverClusterInvoker<T>(directory);
}

    Directory接口：

public interface Directory<T> extends Node {
    //获取接口

   Class<T> getInterface();

    //获取服务列表

    List<Invoker<T>> list(Invocation invocation) throws RpcException;
}

    在Dubbo发起服务调用时，所有理由和负载均衡都是在客户端实现的，如上图步骤②③④所示。客户端服务调用首先触发路由操作，然后将路由结果得到的服务列表作为负载均衡参数，经过负载均衡后会选出一台机器进行rpc调用，会将请求交给底层I/O线程池（比如netty）处理，线程池中主要处理读写、序列化、反序列化等逻辑。在处理反序列化对象时会在业务线程池中处理。在步骤⑤中包含两种线程池，一种是I/0线程池-netty，另一种是Dubbo业务线程池。

    最后服务提供方会根据传递过来的接口、分组和版本查找invoker对应的实例进行反射调用。

二、服务提供方的查找和调用

    服务提供方查询服务方法和调用主要实现在ExchangeHandlerAdapter中，内部实现代码：

public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
    if (message instanceof Invocation) {
        Invocation inv = (Invocation) message;      
        //从exportedMap缓存中获取invoker
        Invoker<?> invoker = getInvoker(channel, inv);

        //...

        //调用业务方具体方法

        Result result = invoker.invoke(inv);
        if (result instanceof AsyncRpcResult) {
            return ((AsyncRpcResult) result).getResultFuture().thenApply(r -> (Object) r);
        } else {
            return CompletableFuture.completedFuture(result);
        }
    }
    //...
}

Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
 //...
    //根据服务接口名、端口、接口分组、版本构建缓存的key
    String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));
 //从缓存中获取exporter对象
    DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
    if (exporter == null)
        throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + inv);
 //返回invoker对象
    return exporter.getInvoker();
}

    服务提供方查询服务方法和调用的源码比较简单，这里应用了exporterMap，这个map在之前服务暴露中提到过，服务暴露时会把暴露的服务根据特点的规则生成构造key-服务接口名+端口+分组+版本，value为服务被包装成的exporter对象。

    因此在服务查找中，也会根据服务接口名+端口+分组+版本进行key的构建，从缓存中查找exporter对象，然后通过getInvoer方法找到invoker对象。最后通过invoker.invoke调用服务的具体业务逻辑。

三、Dubbo的请求响应的处理

    Dubbo框架内部，所有方法的调用会被抽象成Request/Response，即每一次调用都会创建一个Request请求，方法的返回则会返回一个Response对象。请求/响应的处理主要在HeaderExchangeHandler中。

    请求响应实现代码：

public void received(Channel channel, Object message) throws RemotingException {
    //更新事件时间戳
    channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
    final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
    try {
        if (message instanceof Request) {
            Request request = (Request) message;
 if (request.isEvent()) {
                //如果是只读事件，在channel中打标
                handlerEvent(channel, request);
            } else {
                if (request.isTwoWay()) {
                    //处理方法调用并返回
                    handleRequest(exchangeChannel, request);
                } else {
                    handler.received(exchangeChannel, request.getData());
                }
            }
        } else if (message instanceof Response) {
            //接收响应
            handleResponse(channel, (Response) message);
 } else if (message instanceof String) {
            //客户端不能进行telnet调用
            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 {
                //触发telnet调用
                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);
    }
}
//请求处理
void handleRequest(final ExchangeChannel channel, Request req) throws RemotingException {
    Response res = new Response(req.getId(), req.getVersion());
    if (req.isBroken()) {
        Object data = req.getData();
        String msg;
 if (data == null) msg = null;
        //处理请求格式不正确，把异常转换字符串返回
        else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
        else msg = data.toString();
        res.setErrorMessage("Fail to decode request due to: " + msg);
        res.setStatus(Response.BAD_REQUEST);
        channel.send(res);
        return;
    }
    // find handler by message class.
    Object msg = req.getData();
    try {
        //调用上述提到过的DubboProtocol.reply方法进行方法调用
        CompletableFuture<Object> future = handler.reply(channel, msg);
        if (future.isDone()) {
            res.setStatus(Response.OK);
            res.setResult(future.get());
            channel.send(res);
            return;
        }
        //...
    } catch (Throwable e) {
        res.setStatus(Response.SERVICE_ERROR);
        res.setErrorMessage(StringUtils.toString(e));
        channel.send(res);
    }
}
//响应处理
static void handleResponse(Channel channel, Response response) throws RemotingException {
if (response != null && !response.isHeartbeat()) {
        //唤醒阻塞的线程并通知结果
        DefaultFuture.received(channel, response);
    }
}

    上述代码中首先会更新时间戳，然后判断是请求模式还是响应模式还是telnet调用模式，进行相应处理。

    请求模式，主要处理只读事件，用于Dubbo优雅停机。当注册中心反注册元数据时，因为网络问题，客户端不能及时感应注册中心事件，服务端会发送readonly报文告知下线。如果不是只读事件，则调用Dubboprotocol.reply方法进行方法调用。

    响应模式，唤醒阻塞的线程并告知业务调用方。

    telnet模式，触发telnet调用。

四、总结

    本章偏理论，原理也容易懂。主要分析了Dubbo调用原理和流程。后面也分析了服务提供方的查找和方法调用、dubbo的请求/响应处理。

参考资料：

    《深入理解Apache Dubbo与实战》 - 第6章 Dubbo远程调用