文档章节

dubbo 超时设置和源码分析

我叫董先森
 我叫董先森
发布于 09/13 18:16
字数 2301
阅读 403
收藏 11

本文 dubbo 2.6.2

在工作中碰到一个业务接口时间比较长,需要修改超时时间,不知道原理,在网上搜索,看到有人说如果你觉得自己了解了dubbo的超时机制,那么问问自己以下问题:

  • 超时是针对消费端还是服务端?
  • 超时在哪设置?
  • 超时设置的优先级是什么?
  • 超时的实现原理是什么?
  • 超时解决的是什么问题 ?

如果连这些都回答不上了,那只能说明还没有完全掌握 dubbo的超时机制。

于是索性就自己本地搭了个环境,研究了一下源码。 先来说一说结论:

  1. 超时是针对消费端的,消费端会抛出TimeoutException 而服务器端仅仅是一个 warn日志
  2. 超时在消费端、服务器端设置,dubbo会合并这两个设置
  3. consumer方法级别 > provider 方法级别 > consumer 接口级别 > provider 接口级别 > consumer 全局级别 > provider 全局级别。如果都没配置,那么就是dubbo默认的1秒
  4. 见下面分析
  5. 最主要是宝贵的线程,客户端的用户线程不能因为服务端超时而一直类似wait, 导致无法正常响应其他业务。

一、超时时间设置

全局超时配置
<dubbo:consumer timeout="5000" />
指定接口以及特定方法超时配置
    <dubbo:service interface="me.kimi.samples.dubbo.facade.QuestionFacade" ref="questionFacade" timeout="6000">
        <dubbo:method name="getQuestionById" timeout="7000"/>
    </dubbo:service>

观察控制台打印的注册URL:

consumer://172.16.71.30/me.kimi.samples.dubbo.facade.QuestionFacade?application=demo-consumer&category=providers,configurators,routers&check=false&default.proxy=jdk&default.timeout=5000&dubbo=2.6.2&getQuestionById.timeout=7000&interface=me.kimi.samples.dubbo.facade.QuestionFacade&logger=log4j&methods=getQuestionById&pid=13884&side=consumer&timeout=6000&timestamp=1536630294523

可以看到:

  • default.timeout=5000
  • timeout=6000
  • getQuestionById.timeout=7000

分别对应了全局、类级别、方法级别的超时设置。

省略一部分调用链,最终会来到这里 DubboInvoker,读取超时时间:

com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker

    @Override
    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);
            // 读取超时时间,这里dubbo已经把服务端的timeout参数和消费端的timeout参数合并
            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);
                // 返回 DefaultFuture
                // get()在没返回值之前会 阻塞 await
                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);
        }
    }

看一下参数获取的方法:

public int getMethodParameter(String method, String key, int defaultValue) {
    // 首先查 getQuestionById.timeout
    String methodKey = method + "." + key;
    // 从数字缓存中先获取,不需要每次都 parseInt
    Number n = getNumbers().get(methodKey);
    if (n != null) {
        return n.intValue();
    }
    // 没得话,去取字符串值
    String value = getMethodParameter(method, key);
    if (value == null || value.length() == 0) {
        // 三个地方都没配置,返回默认值,默认是1秒
        return defaultValue;
    }
    // 放入缓存中
    int i = Integer.parseInt(value);
    getNumbers().put(methodKey, i);
    return i;
}
public String getMethodParameter(String method, String key) {
    // 首先查 getQuestionById.timeout
    String value = parameters.get(method + "." + key);
    if (value == null || value.length() == 0) {
        // 没有设定方法级别的,去查接口级别或全局的
        return getParameter(key);
    }
    return value;
}
public String getParameter(String key) {
    // 接口级别去查 timeout 
    String value = parameters.get(key);
    if (value == null || value.length() == 0) {
        // 没的话查询全局级别 default.timeout
        value = parameters.get(Constants.DEFAULT_KEY_PREFIX + key);
    }
    return value;
}

从代码中可以看出超时时间的设置:方法级别 > 接口级别 > 全局级别。

这里要特殊提一点,就是dubbo会合并服务端客户端的设置。

修改客户端配置, 只留下全局设置:

<dubbo:consumer timeout="2000" proxy="jdk"/>
    <dubbo:service interface="me.kimi.samples.dubbo.facade.QuestionFacade" ref="questionFacade"/>

服务端配置如下:

    <dubbo:provider timeout="10000" accepts="500"/>
    <!-- service implementation, as same as regular local bean -->
    <bean id="questionFacade" class="me.kimi.samples.dubbo.provider.service.QuestionFacadeImpl"/>
    <!-- declare the service interface to be exported -->
    <dubbo:service interface="me.kimi.samples.dubbo.facade.QuestionFacade" ref="questionFacade" timeout="9000"/>

最后在客户端调用的时候,发现timeout是9000ms, debug发现客户端合并了url, 合并结果如下:

dubbo://172.16.71.30:20880/me.kimi.samples.dubbo.facade.QuestionFacade?anyhost=true&application=demo-provider&default.accepts=500&default.timeout=10000&dubbo=2.6.2&generic=false&interface=me.kimi.samples.dubbo.facade.QuestionFacade&logger=log4j&methods=getQuestionById&pid=17508&side=provider&timeout=9000&timestamp=1536660132286

查看源码 com.alibaba.dubbo.registry.integration.RegistryDirectory#mergeUrl:

private URL mergeUrl(URL providerUrl) {
    providerUrl = ClusterUtils.mergeUrl(providerUrl, queryMap); // Merge the consumer side parameters

    List<Configurator> localConfigurators = this.configurators; // local reference
    if (localConfigurators != null && !localConfigurators.isEmpty()) {
        for (Configurator configurator : localConfigurators) {
            providerUrl = configurator.configure(providerUrl);
        }
    }

    providerUrl = providerUrl.addParameter(Constants.CHECK_KEY, String.valueOf(false)); // Do not check whether the connection is successful or not, always create Invoker!

    // 这里就是合并服务器端的参数,所以除了timeout参数,其他很多参数也是这样的
    // 即已客户端优先
    this.overrideDirectoryUrl = this.overrideDirectoryUrl.addParametersIfAbsent(providerUrl.getParameters()); 

    if ((providerUrl.getPath() == null || providerUrl.getPath().length() == 0)
            && "dubbo".equals(providerUrl.getProtocol())) { // Compatible version 1.0
        //fix by tony.chenl DUBBO-44
        String path = directoryUrl.getParameter(Constants.INTERFACE_KEY);
        if (path != null) {
            int i = path.indexOf('/');
            if (i >= 0) {
                path = path.substring(i + 1);
            }
            i = path.lastIndexOf(':');
            if (i >= 0) {
                path = path.substring(0, i);
            }
            providerUrl = providerUrl.setPath(path);
        }
    }
    return providerUrl;
}

所以综合,超时时间的优先级为:

consumer方法级别 > provider 方法级别 > consumer 接口级别 > provider 接口级别 > consumer 全局级别 > provider 全局级别。

二、超时实现

有了超时时间,那么dubbo是怎么实现超时的呢?

再看上面的DubboInvoker,对于一般的有返回值的调用,最终调用:

return (Result) currentClient.request(inv, timeout).get();

先看一下request方法,来到 com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeChannel的Request方法:

    @Override
    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("2.0.0");
        req.setTwoWay(true);
        req.setData(request);
       
        DefaultFuture future = new DefaultFuture(channel, req, timeout);
        try {
            channel.send(req);
        } catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }

重点是 DefaultFuture:

static {
    Thread th = new Thread(new RemotingInvocationTimeoutScan(), "DubboResponseTimeoutScanTimer");
    th.setDaemon(true);
    th.start();
}

类加载的时候会启动一个超时扫描线程:

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);
    // 每个 DefaultFuture 都有一个 id, 对应当前请求id, 然后被放到 静态Map中。
    FUTURES.put(id, this);
    // id 对应的 Channel 也存起来,后续超时需要处理
    CHANNELS.put(id, channel);
}

再看下get方法:

@Override
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();
}

从上面代码上可以看到,get方法,会使当前线程挂起等待。那么什么时候会被恢复呢,可以想到两类情况:

  1. 超时
  2. 服务端正常返回

那么回过头来看看超时扫描线程,看一下扫描线程做了什么事情:

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;
                        }
                        // 如果future未完成,且超时
                        if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                            // 创建一个异常的Response
                            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(future.getChannel(), timeoutResponse);
                        }
                    }
                    Thread.sleep(30);
                } catch (Throwable e) {
                    logger.error("Exception when scan the timeout invocation of remoting.", e);
                }
            }
        }
    }

看下 received方法

public static void received(Channel channel, Response response) {
    try {
        DefaultFuture future = FUTURES.remove(response.getId());
        if (future != null) {
            future.doReceived(response);
        } else {
            logger.warn("The timeout response finally returned at "
                    + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
                    + ", response " + response
                    + (channel == null ? "" : ", channel: " + channel.getLocalAddress()
                    + " -> " + channel.getRemoteAddress()));
        }
    } finally {
        CHANNELS.remove(response.getId());
    }
}
private void doReceived(Response res) {
    lock.lock();
    try {
        // 设置响应
        // 这样isDone就是true了
        response = res;
        if (done != null) {
            // 恢复挂起的线程 
            done.signal();
        }
    } finally {
        lock.unlock();
    }
    if (callback != null) {
        invokeCallback(callback);
    }
}

显然这里扫描线程把用户请求线程恢复了。 恢复以后,顺着刚才的 DefaultFuture 的get方法,来到 returnFromResponse方法:

private Object returnFromResponse() throws RemotingException {
    Response res = response;
    if (res == null) {
        throw new IllegalStateException("response cannot be null");
    }
    // 正常返回,返回 Result 对象
    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());
}

超时扫描线程,构建了一个 超时 Response, 在这里抛出 超时异常。

超时抛异常是看见了,那么正常返回是怎么处理的呢,因为 done还 await在那里。 这里暂时不细说dubbo其他组件的原理,只要知道在网络事件完成(即服务器端在规定时间内正常返回)的时候,会有个回调,在整个回调过程中,最终会回调到 com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler 的 received 方法,看下代码:

@Override
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);
    }
}

处理响应:

static void handleResponse(Channel channel, Response response) throws RemotingException {
    // 不是心跳包,是正常的业务返回
    if (response != null && !response.isHeartbeat()) {
        DefaultFuture.received(channel, response);
    }
}

这里看到,最终调用也是 DefaultFuture.received 的方法,和超时扫描的入口一样, 最终会恢复用户请求线程。唯一有区别的就是,这里是一个ok的Response, 而那边是timeout的response.

© 著作权归作者所有

共有 人打赏支持
我叫董先森
粉丝 2
博文 24
码字总数 19433
作品 0
杭州
dubbo超时异常

dubbo超时异常 在调用dubbo服务时经常看到如下错误:Caused by: com.alibaba.dubbo.remoting.TimeoutException: Waiting server-side response timeout by scan timer. 源码分析 客户端调用远......

lipengHeke
02/05
13
0
使用Dubbo中需要注意的事项

一、前言 Dubbo作为高性能RPC框架,已经进入Apache卵化器项目,虽然官方给出了dubbo使用的用户手册,但是大多是一概而过,使用dubbo时候要尽量了解源码,不然会很容易入坑。 二 、服务消费端...

加多
01/02
0
0
Dubbo 实现原理与源码解析系列 —— 精品合集

摘要: 原创出处 http://www.iocoder.cn/Dubbo/good-collection/ 「芋道源码」欢迎转载,保留摘要,谢谢! 1.【芋艿】精尽 Dubbo 原理与源码专栏 2.【老徐】RPC 专栏 3.【肥朝】Dubbo 源码解析...

芋道源码掘金Java群217878901
06/23
0
0
Dubbo 服务调用 源码学习(下)(七)

笔记简述 本学习笔记接上篇Dubbo 服务调用 源码(上)学习(六),上一篇已经完成了invoker的生成,接下来就是具体的方法调用了,包含了mock测试、负载均衡(不涉及细节)、重试、netty调用、...

jwfy
05/16
0
0
架构师必备词汇和知识点

01 高可用 负载均衡(负载均衡算法) 反向代理 服务隔离 服务限流 服务降级(自动优雅降级) 失效转移 超时重试(代理超时、容器超时、前端超时、中间件超时、数据库超时、NoSql超时) 回滚机...

t4i2b10X4c22nF6A
2017/11/24
0
0

没有更多内容

加载失败,请刷新页面

加载更多

xilinx资源

本系列教学视频由赛灵思高级战略应用工程师带领你:从零开始,一步步深入 掌握 HLS 以及 UltraFAST 设计方法,帮助您成为系统设计和算法加速的大拿! http://www.eetrend.com/topics/2018-0...

whoisliang
25分钟前
0
0
=====BJmeter性能测试小接=====

一、性能测试分类 1、负载测试: 通过逐步加压的方法,达到既定的性能阈值的目标,阈值的设定应是小于某个值,如cpu使用率小于等于80% 2、压力测试: 通过逐步加压的方法,使得系统的某些资源...

覃光林
29分钟前
1
0
企业级开源四层负载均衡解决方案--LVS

网盘链接 企业级开源四层负载均衡解决方案--LVS 本课程将在Linux环境下,学习配置使用LVS,对Web集群和MySQL集群进行负载均衡,并结合利用Keepalived实现负载均衡器的高可用,实现对后端Rea...

qq__2304636824
34分钟前
2
0
Windows上安装Spacemacs

emacs安装 下载地址emacs 安装比较简单,解压后执行\bin\addpm.exe即可 emacs配置 emacs的默认配置文件路径和.emacs.d文件夹都是在Windows主目录下的 C:\Users\Administrator\AppData\Roami...

yxmsw2007
50分钟前
0
0
OSChina 周一乱弹 —— 鱼生不值得

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @瘟神灬念:分享新裤子的单曲《没有理想的人不伤心 (Remix版)》: 《没有理想的人不伤心 (Remix版)》- 新裤子 手机党少年们想听歌,请使劲儿戳...

小小编辑
今天
227
9

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部