dubbo超时机制原理

在dubbo中，可以设置服务的超时时间，服务端和消费端可以分别设置，但是服务消费端在调用的时候，会根据timeout配置的优先级，去获取阈值，消费端的配置高于消费端的，那这篇笔记主要记录的是服务消费端超时的原理，或者说，是服务调用方是如何判断服务调用时间超过了设置的阈值

结论

1.首先服务调用方，会根据配置优先级，获取一个超时时间(消费端配置 > 服务提供端配置接口层面 > 应用层面配置)
2.然后会把超时时间一直进行透传，通过invocation对象
3.然后在调用端通过netty发送了请求之后，会通过Condition.await()来等待指定的时间
3.1 如果休眠指定的时间之后，调用执行完毕，就返回
3.2 如果依旧没有拿到返回信息，就抛出timeout异常

所以：我认为，简单而已，dubbo的超时机制，就是通过condition同步条件队列来完成的

源码阅读

在发起一笔调用的时候，调用链是这样的

com.alibaba.dubbo.rpc.protocol.AbstractInvoker#invoke
com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
调用链是上面这个，就是服务消费端在调用的时候，会通过netty进行服务调用，netty客户端会在服务引入的时候，提前创建好，所以在dubboInvoker的doInvoke()方法中，会直接根据url去集合中取出来一个client，发起请求


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);// 这里是获取一个netty服务端，通过exchangeClient去发起netty请求调用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);/*** 在发起调用时，获取超时时间的源码* 1.先取方法上的配置，如果方法上未配置* 2.接着取接口上的配置* 3.如果接口上也未配置，那就取默认的配置** 总之，这里的timeout是根据优先级过滤之后，得到的超时时间*/int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);/*** 1.如果是单路的，就无需异步或者同步，在调用之后，直接return即可* 2.如果是异步调用，就通过future获取到返回结果* 3.如果是同步调用，就等待返回结果*/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);/*** dubbo服务调用者超时的判断，就在下面这行代码中*/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);}
}

这里比较重要的是：currentClient.request()这个方法和get()方法
我们以同步调用为例

request()

这里可以看到，把超时时间放到的defaultFuture对象中，其实这里放到这个对象中，就是为了在发起调用之后，去通过condition.await()来休眠指定的时间

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);// 这里会把超时时间放到future中，然后通过channel.send()去发放真正的netty请求DefaultFuture future = new DefaultFuture(channel, req, timeout);try {channel.send(req);} catch (RemotingException e) {future.cancel();throw e;}return future;
}

放到defaultFuture对象中，一定是为了在后面使用的

get()


public Object get(int timeout) throws RemotingException {if (timeout <= 0) {timeout = Constants.DEFAULT_TIMEOUT;}/*** isDone()就是判断服务提供者是否执行完毕，通过response是否为null来判断* 如果已经执行完毕，就不会进入到下面if判断中* 如果未执行完毕*  1.加锁，reentrantLock*  2.然后双重检查，再判断服务提供者是否执行完毕*  3.如果没有执行完毕，就调用condition.await()，将当前线程放入到同步条件队列中，在指定的timeout时间到了之后，线程会继续执行*  4.再次判断是否执行完毕，如果执行完毕或者 当前时间 - start 超过了timeout，就表示要么超时了，要么执行完毕了*  5.最后会判断，如果是执行完毕了，就return即可，如果是未执行完毕,就抛出timeOutException*/if (!isDone()) {long start = System.currentTimeMillis();lock.lock();try {while (!isDone()) {/*** 如果没有执行完毕，调用condition的await方法休眠指定的时间* 然后判断是否执行完毕*/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();
}

总和以上三部分代码，我们可以发现这样一个逻辑
1.首先会判断是异步调用、还是同步调用，无论是异步还是同步，都会发起一个netty请求，如果是异步的话，会直接返回一个RpcResult对象，在后面的某一个流程中，去调用defaultFuture的get()方法
2.发起请求时，会先把超时时间放到defaultFuture对象中，因为ExchangeClient.request()方法，返回的是defaultFuture对象
3.如果是同步请求，就调用defaultFuture的get()方法阻塞，直到请求正常返回；如果超过了timeout，服务提供者依然没有返回数据，就抛出超时异常

condition

无论是异步调用，还是同步调用，都是通过future.get()，或者说是AQS中的condition来实现的
这里就是jdk的源码了，在调用condition.await()方法的时候，会把当前线程放入到一个同步条件队列中，同时，将当前线程从同步等待队列中remove
在指定的timeout时间到了之后，将线程从同步条件对列中移到同步等待队列中
进入到同步等待队列中之后，线程就会继续接着执行

服务端如果超时了会怎么处理？

举个例子：
dubbo消费者服务超时时间是5S，dubbo服务端配置的超时时间是2S,但是服务端代码的执行，实际需要3S，那这种情况下
服务消费者在调用的时候，会以5S为超时时间，而服务端还是以2S为超时时间
经过测试，发现这种场景下，服务消费端可以正常获取到返回结果，服务提供者会打印一个超时warn日志，并不会抛出异常
具体实现类是在：com.alibaba.dubbo.rpc.filter.TimeoutFilter中

/*** Log any invocation timeout, but don't stop server from running* 超时判断，只作用于服务提供者*/
@Activate(group = Constants.PROVIDER)
public class TimeoutFilter implements Filter {private static final Logger logger = LoggerFactory.getLogger(TimeoutFilter.class);@Overridepublic Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {// 1.记录开始时间long start = System.currentTimeMillis();// 2.调用服务端代码Result result = invoker.invoke(invocation);// 3.记录服务端代码执行时间long elapsed = System.currentTimeMillis() - start;// 4.如果执行时间超过了timeout配置，就打印warn日志，然后returnif (invoker.getUrl() != null&& elapsed > invoker.getUrl().getMethodParameter(invocation.getMethodName(),"timeout", Integer.MAX_VALUE)) {if (logger.isWarnEnabled()) {logger.warn("invoke time out. method: " + invocation.getMethodName()+ " arguments: " + Arrays.toString(invocation.getArguments()) + " , url is "+ invoker.getUrl() + ", invoke elapsed " + elapsed + " ms.");}}return result;}}