dubbo（九）：timeout超时机制解析
在网络请求时，总会有各种异常情况出现，我们需要提前处理这种情况。在完善的rpc组件dubbo中，自然是不会少了这一层东西的。我们只需要通过一些简单的配置就可以达到超时限制的作用了。
dubbo的设计理念是，客户端控制优先，服务端控制兜底。
1. 超时机制的实现思路
要想实现超时，一般有两个思路。一个是客户端自行设置一个超时限制，达到超时时间还未返回，则抛出异常，不再等待结果。二是通过在超时后，将连接断开，从而使数据请求中断，最终也是以抛出异常的方式返回的。
当然，超时有两种情况，一种是自己主动的超时，另一种是被别人关掉连接发生的超时（需主动主发送超时消息）。一般我们认为主动设置的超时是可控的，被动的超时将是一个不可逾越的鸿沟，如果必须需要更长的时间才能拿到结果时，此种超时将限制我们，我们只能另谋出路了，比如调用的异步化。
一般地，要想实现超时，我们也有两种方式：一种是调用别人提供的api，其中包含了超时设置，此时仅需简单设置即可；另一种是我们自行实现的超时，比如原本只有一个无限接口，我们要实现超时，必须将其异步化，通过额外的线程来进行超时的检测和控制。
那么，dubbo又是怎样实现超时的呢？
2. 客户端实现超时
我们前面说过，dubbo中consumer端可以设置超时，服务端也可以提供超时设置。那么，会不会是客户端和服务端都要实现超时机制呢？不管怎么样，客户端是一定要做的。所以，我们先来看看客户端实现超时的机制。
2.1. 客户端使用超时的方式
首先，dubbo的调置超时方式，按照其整体架构设计理念，都有几个作用域：应用级 -> 接口级 -> 方法级。 consumer端 -> provider端。
// 消费者端特定方法的配置 <dubbo:consumer interface="com.alibaba.xxx.XxxService" > <dubbo:method name="findPerson" timeout="1000" /> </dubbo:consumer> // 消费者端特定接口的配置 <dubbo:consumer interface="com.alibaba.xxx.XxxService" timeout="200" /> // 提供者端特定方法的配置 <dubbo:service interface="com.alibaba.xxx.XxxService" > <dubbo:method name="findPerson" timeout="1000" /> </dubbo:service> // 提供者端特定接口的配置 <dubbo:service interface="com.alibaba.xxx.XxxService" timeout="200" />
当然了，上面这种是使用xml进行配置的，你还可以使用properties文件进行配置，也可以使用java代码直接进行配置。
2.2. 超时参数的读取与使用
这些参数设置好后，在调用rpc时，进行想入相应的Invoket，进行读取参数，使用。
// org.apache.dubbo.rpc.protocol.AsyncToSyncInvoker#invoke @Override public Result invoke(Invocation invocation) throws RpcException { // 同步和异步，底层都是异步请求，仅做上层封装 Result asyncResult = invoker.invoke(invocation); try { // 同步请求时，在内部等待 if (InvokeMode.SYNC == ((RpcInvocation) invocation).getInvokeMode()) { /** * NOTICE! * must call {@link java.util.concurrent.CompletableFuture#get(long, TimeUnit)} because * {@link java.util.concurrent.CompletableFuture#get()} was proved to have serious performance drop. */ asyncResult.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS); } } catch (InterruptedException e) { throw new RpcException("Interrupted unexpectedly while waiting for remote result to return! method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } catch (ExecutionException e) { Throwable t = e.getCause(); // 超时返回，给出详细堆栈 if (t instanceof TimeoutException) { throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } else if (t instanceof RemotingException) { throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } else { throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } } catch (Throwable e) { throw new RpcException(e.getMessage(), e); } return asyncResult; } // org.apache.dubbo.rpc.protocol.AbstractInvoker#invoke @Override public Result invoke(Invocation inv) throws RpcException { // if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed if (destroyed.get()) { logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, " + ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer"); } RpcInvocation invocation = (RpcInvocation) inv; invocation.setInvoker(this); if (CollectionUtils.isNotEmptyMap(attachment)) { invocation.addObjectAttachmentsIfAbsent(attachment); } Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments(); if (CollectionUtils.isNotEmptyMap(contextAttachments)) { /** * invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here, * because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered * by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is * a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information). */ invocation.addObjectAttachments(contextAttachments); } invocation.setInvokeMode(RpcUtils.getInvokeMode(url, invocation)); RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation); Byte serializationId = CodecSupport.getIDByName(getUrl().getParameter(SERIALIZATION_KEY, DEFAULT_REMOTING_SERIALIZATION)); if (serializationId != null) { invocation.put(SERIALIZATION_ID_KEY, serializationId); } AsyncRpcResult asyncResult; try { // 调用远程方法 asyncResult = (AsyncRpcResult) doInvoke(invocation); } catch (InvocationTargetException e) { // biz exception Throwable te = e.getTargetException(); if (te == null) { asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation); } else { if (te instanceof RpcException) { ((RpcException) te).setCode(RpcException.BIZ_EXCEPTION); } asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, te, invocation); } } catch (RpcException e) { if (e.isBiz()) { asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation); } else { throw e; } } catch (Throwable e) { asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation); } RpcContext.getContext().setFuture(new FutureAdapter(asyncResult.getResponseFuture())); return asyncResult; } // org.apache.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke @Override protected Result doInvoke(final Invocation invocation) throws Throwable { RpcInvocation inv = (RpcInvocation) invocation; final String methodName = RpcUtils.getMethodName(invocation); inv.setAttachment(PATH_KEY, getUrl().getPath()); inv.setAttachment(VERSION_KEY, version); ExchangeClient currentClient; if (clients.length == 1) { currentClient = clients[0]; } else { currentClient = clients[index.getAndIncrement() % clients.length]; } try { boolean isOneway = RpcUtils.isOneway(getUrl(), invocation); // 获取超时设置 int timeout = calculateTimeout(invocation, methodName); invocation.put(TIMEOUT_KEY, timeout); if (isOneway) { boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false); currentClient.send(inv, isSent); return AsyncRpcResult.newDefaultAsyncResult(invocation); } else { // 响应结果回调，使用线程池接收 ExecutorService executor = getCallbackExecutor(getUrl(), inv); // 向服务端发送请求，并返回 future 作为结果接收器 CompletableFuture<AppResponse> appResponseFuture = currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj); // save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter FutureContext.getContext().setCompatibleFuture(appResponseFuture); AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv); result.setExecutor(executor); return result; } } 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); } } // 超时配置读取，多种方式，多种优先级 private int calculateTimeout(Invocation invocation, String methodName) { // timeout-countdown, 需要传导到服务端的超时控制 Object countdown = RpcContext.getContext().get(TIME_COUNTDOWN_KEY); // 默认1s超时 int timeout = DEFAULT_TIMEOUT; if (countdown == null) { timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getContext(), DEFAULT_TIMEOUT); if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) { invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout); // pass timeout to remote server } } else { TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countdown; timeout = (int) timeoutCountDown.timeRemaining(TimeUnit.MILLISECONDS); invocation.setObjectAttachment(TIMEOUT_ATTACHMENT_KEY, timeout);// pass timeout to remote server } return timeout; } // org.apache.dubbo.rpc.support.RpcUtils#getTimeout public static long getTimeout(URL url, String methodName, RpcContext context, long defaultTimeout) { long timeout = defaultTimeout; // 先方法，后接口 // 事实上，所有接口级的变量在注册的时候已经作用到了方法级上了，所以只需读取方法上的参数即可 Object genericTimeout = context.getObjectAttachment(TIMEOUT_KEY); if (genericTimeout != null) { timeout = convertToNumber(genericTimeout, defaultTimeout); } else if (url != null) { timeout = url.getMethodPositiveParameter(methodName, TIMEOUT_KEY, defaultTimeout); } return timeout; } // org.apache.dubbo.common.URL#getMethodPositiveParameter(java.lang.String, java.lang.String, long) public long getMethodPositiveParameter(String method, String key, long defaultValue) { if (defaultValue <= 0) { throw new IllegalArgumentException("defaultValue <= 0"); } long value = getMethodParameter(method, key, defaultValue); return value <= 0 ? defaultValue : value; } public long getMethodPositiveParameter(String method, String key, long defaultValue) { if (defaultValue <= 0) { throw new IllegalArgumentException("defaultValue <= 0"); } long value = getMethodParameter(method, key, defaultValue); return value <= 0 ? defaultValue : value; } // org.apache.dubbo.common.URL#getMethodParameter(java.lang.String, java.lang.String, long) public long getMethodParameter(String method, String key, long defaultValue) { Number n = getCachedNumber(method, key); if (n != null) { return n.longValue(); } String value = getMethodParameter(method, key); if (StringUtils.isEmpty(value)) { return defaultValue; } long l = Long.parseLong(value); updateCachedNumber(method, key, l); return l; } // org.apache.dubbo.rpc.protocol.AbstractInvoker#getCallbackExecutor protected ExecutorService getCallbackExecutor(URL url, Invocation inv) { ExecutorService sharedExecutor = ExtensionLoader.getExtensionLoader(ExecutorRepository.class).getDefaultExtension().getExecutor(url); if (InvokeMode.SYNC == RpcUtils.getInvokeMode(getUrl(), inv)) { // 同步请求使用少量的共享线程池，实际上是做进一步封装处理 return new ThreadlessExecutor(sharedExecutor); } else { // 异步调用则直接使用共享线程池，不受其他节点控制 return sharedExecutor; } }
从上面可以看出，dubbo的超时机制是通过异步线程future的方式实现的，其中，同步调用的超时设置，底层也是用异步实现。这样既简化了底层实现，也对外提供了很好的易用性。因为底层都是通过netty或nio实现网络通信，而这种实现一般又是select-poll 模型或者 epoll 模型，反正也必须要用异步处理，所以不管如何也是跑不掉这个实现。只要实现好一个底层异步通知，全部基石就都好了。而上层，则只需关注是用户实现，还是框架实现了。
2.3. 客户端超时监控处理
上面的实现，我们并没有看到具体是如何实现超时的，毕竟我们只是看到了表面现象，即只是设置了一个 timeout参数，而已。更深层次的实现，请继续。也就是说dubbo是在做请求的同时，做了超时的设置工作。
// org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeChannel#request @Override public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor) 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); // 里面包含了一个超时任务 timeTask DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout, executor); try { channel.send(req); } catch (RemotingException e) { future.cancel(); throw e; } return future; } // org.apache.dubbo.remoting.exchange.support.DefaultFuture#newFuture /** * init a DefaultFuture * 1.init a DefaultFuture * 2.timeout check * * @param channel channel * @param request the request * @param timeout timeout * @return a new DefaultFuture */ public static DefaultFuture newFuture(Channel channel, Request request, int timeout, ExecutorService executor) { final DefaultFuture future = new DefaultFuture(channel, request, timeout); future.setExecutor(executor); // ThreadlessExecutor needs to hold the waiting future in case of circuit return. if (executor instanceof ThreadlessExecutor) { ((ThreadlessExecutor) executor).setWaitingFuture(future); } // timeout check timeoutCheck(future); return future; } // org.apache.dubbo.remoting.exchange.support.DefaultFuture#timeoutCheck /** * check time out of the future */ private static void timeoutCheck(DefaultFuture future) { TimeoutCheckTask task = new TimeoutCheckTask(future.getId()); // 添加一个定时器 future.timeoutCheckTask = TIME_OUT_TIMER.newTimeout(task, future.getTimeout(), TimeUnit.MILLISECONDS); } // org.apache.dubbo.common.timer.HashedWheelTimer#newTimeout @Override public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) { if (task == null) { throw new NullPointerException("task"); } if (unit == null) { throw new NullPointerException("unit"); } long pendingTimeoutsCount = pendingTimeouts.incrementAndGet(); if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) { pendingTimeouts.decrementAndGet(); throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending " + "timeouts (" + maxPendingTimeouts + ")"); } start(); // Add the timeout to the timeout queue which will be processed on the next tick. // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket. long deadline = System.nanoTime() + unit.toNanos(delay) - startTime; // Guard against overflow. if (delay > 0 && deadline < 0) { deadline = Long.MAX_VALUE; } HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline); timeouts.add(timeout); return timeout; } // org.apache.dubbo.remoting.exchange.support.DefaultFuture.TimeoutCheckTask#TimeoutCheckTask TimeoutCheckTask(Long requestID) { this.requestID = requestID; } @Override public void run(Timeout timeout) { DefaultFuture future = DefaultFuture.getFuture(requestID); if (future == null || future.isDone()) { return; } if (future.getExecutor() != null) { future.getExecutor().execute(() -> notifyTimeout(future)); } else { notifyTimeout(future); } } // org.apache.dubbo.remoting.exchange.support.DefaultFuture.TimeoutCheckTask#notifyTimeout private void notifyTimeout(DefaultFuture future) { // create exception 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)); // handle response. DefaultFuture.received(future.getChannel(), timeoutResponse, true); } // org.apache.dubbo.remoting.exchange.support.DefaultFuture#received public static void received(Channel channel, Response response, boolean timeout) { try { DefaultFuture future = FUTURES.remove(response.getId()); if (future != null) { Timeout t = future.timeoutCheckTask; if (!timeout) { // decrease Time t.cancel(); } 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 status is " + response.getStatus() + (channel == null ? "" : ", channel: " + channel.getLocalAddress() + " -> " + channel.getRemoteAddress()) + ", please check provider side for detailed result."); } } finally { CHANNELS.remove(response.getId()); } } // 抛出异常消息 private void doReceived(Response res) { if (res == null) { throw new IllegalStateException("response cannot be null"); } if (res.getStatus() == Response.OK) { this.complete(res.getResult()); } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) { // 封装为 TimeoutException this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage())); } else { this.completeExceptionally(new RemotingException(channel, res.getErrorMessage())); } // the result is returning, but the caller thread may still waiting // to avoid endless waiting for whatever reason, notify caller thread to return. if (executor != null && executor instanceof ThreadlessExecutor) { ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor; if (threadlessExecutor.isWaiting()) { threadlessExecutor.notifyReturn(new IllegalStateException("The result has returned, but the biz thread is still waiting" + " which is not an expected state, interrupt the thread manually by returning an exception.")); } } }
总体来说就是，在提交服务端的查询请求时，会开启定时任务，检查超时。如果定时任务到期，还未收到结果则会触发超时通知。如果客户端还未成功发送数据，则认为是客户端自己超时。如果已经将数据发送出去，则认为暗服务端超时。这相当于是一个看门狗的形式处理了，就是说，不管服务端和客户端本身如何，总能被这东西给发现，所以这种超时控制是精确的。
当然，除了看门狗的监控，还有的情况是需要应用自己去主动发现的。至少，它不能一直让看门狗起作用吧。
2.4. 异步处理结果的超时处理
异步结果的处理有两种入口方式：一是后台线程处理好之后，自行将结果放置到合适的地方；二是主线程主动查询结果，如果没有完成就等待，直到完成或超时返回；dubbo是在发送请求时，设置一个定时器，检查是否超时，到超时时间就发送一个超时事件。并结束任务。
同步和异步的结果处理方式如下：
// org.apache.dubbo.remoting.transport.dispatcher.all.AllChannelHandler#received @Override public void received(Channel channel, Object message) throws RemotingException { // 根据requestId, 取出之前设定的executor, 提交给业务线程池调用 ExecutorService executor = getPreferredExecutorService(message); try { // 将消息封装成 ChannelEventRunnable, 交由后续处理 executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)); } catch (Throwable t) { if(message instanceof Request && t instanceof RejectedExecutionException){ sendFeedback(channel, (Request) message, t); return; } throw new ExecutionException(message, channel, getClass() + " error when process received event .", t); } } /** * Currently, this method is mainly customized to facilitate the thread model on consumer side. * 1. Use ThreadlessExecutor, aka., delegate callback directly to the thread initiating the call. * 2. Use shared executor to execute the callback. * * @param msg * @return */ public ExecutorService getPreferredExecutorService(Object msg) { if (msg instanceof Response) { Response response = (Response) msg; DefaultFuture responseFuture = DefaultFuture.getFuture(response.getId()); // a typical scenario is the response returned after timeout, the timeout response may has completed the future if (responseFuture == null) { return getSharedExecutorService(); } else { // 取出之前设定的executor ExecutorService executor = responseFuture.getExecutor(); if (executor == null || executor.isShutdown()) { executor = getSharedExecutorService(); } return executor; } } else { return getSharedExecutorService(); } } // 这是同步调用时使用到的线程池 ThreadlessExecutor, 接收到数据后不会立即处理 /** * If the calling thread is still waiting for a callback task, add the task into the blocking queue to wait for schedule. * Otherwise, submit to shared callback executor directly. * * @param runnable */ @Override public void execute(Runnable runnable) { runnable = new RunnableWrapper(runnable); synchronized (lock) { if (!waiting) { sharedExecutor.execute(runnable); } // 只要客户端的还没有触发结果检查，那么将放入队列中，即不会主动进行通知结果 else { queue.add(runnable); } } }
可以看到同步和异步的处理方式区别在于使用不同的线程池实现，异步是直接运行，而同步则做了一次包装，这也为它自定义更合适的处理方式打下了基础。
2.5. 同步请求的结果处理方式
同步处理时，在上层接口调用也是无感的。但是底层都被包装成了异步调用，所以会在上层api中主动进行结果的等待处理。当然，既然是同步处理，它自然是不会主动设置一个较小的超时的，而是用了一个 Integer.MAX_VALUE 的超时设置，真正的超时是由异步结果处理中抛出。
// asyncResult.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS); // org.apache.dubbo.rpc.AsyncRpcResult#get(long, java.util.concurrent.TimeUnit) @Override public Result get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (executor != null && executor instanceof ThreadlessExecutor) { ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor; threadlessExecutor.waitAndDrain(); } // 最终直接从指定位置获取结果即可 return responseFuture.get(timeout, unit); } // org.apache.dubbo.common.threadpool.ThreadlessExecutor#waitAndDrain() /** * Waits until there is a task, executes the task and all queued tasks (if there're any). The task is either a normal * response or a timeout response. */ public void waitAndDrain() throws InterruptedException { /** * Usually, {@link #waitAndDrain()} will only get called once. It blocks for the response for the first time, * once the response (the task) reached and being executed waitAndDrain will return, the whole request process * then finishes. Subsequent calls on {@link #waitAndDrain()} (if there're any) should return immediately. * * There's no need to worry that {@link #finished} is not thread-safe. Checking and updating of * 'finished' only appear in waitAndDrain, since waitAndDrain is binding to one RPC call (one thread), the call * of it is totally sequential. */ if (finished) { return; } Runnable runnable; try { // 如果服务端没有响应，这里是会一直阻塞，因此也达到了同步等待的效果 runnable = queue.take(); }catch (InterruptedException e){ waiting = false; throw e; } // 当拿到结果之后，再运行后续的任务，一般没啥事了，主要就是将结果放置到合适的位置，以后后续可取 synchronized (lock) { waiting = false; runnable.run(); } runnable = queue.poll(); while (runnable != null) { runnable.run(); runnable = queue.poll(); } // mark the status of ThreadlessExecutor as finished. finished = true; }
同步只是表象，异步才是核心。
2.6. 异步的处理方式
异步执行时，使用的就是 ThreadPoolExecutor, 直接进行execute, 即提交到线程池立即执行。即都是统一用共享线程池进行处理，这样做的好处是，不需要等待客户端调用结果，而是主动将结果放置到future的result位置，只需等待处理即可。
// org.apache.dubbo.remoting.transport.dispatcher.ChannelEventRunnable#run @Override public void run() { if (state == ChannelState.RECEIVED) { try { // 直接进入到netty 管道出入站流程，并最终如前面将结果设置到指定位置 handler.received(channel, message); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel + ", message is " + message, e); } } else { switch (state) { case CONNECTED: try { handler.connected(channel); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case DISCONNECTED: try { handler.disconnected(channel); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case SENT: try { handler.sent(channel, message); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel + ", message is " + message, e); } break; case CAUGHT: try { handler.caught(channel, exception); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel + ", message is: " + message + ", exception is " + exception, e); } break; default: logger.warn("unknown state: " + state + ", message is " + message); } } }
异步处理没啥特别的，直接交由netty的pipeline机制完全处理即可。
2.7. netty相关的一点设置
NettypClient, 与服务端交互的入口。主要用于开启网络连接，设置各种处理器，总体来说就是netty的编程模型。感兴趣的自行翻阅。
// org.apache.dubbo.remoting.transport.netty.NettyClient#NettyClient public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException { super(url, wrapChannelHandler(url, handler)); } @Override protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); bootstrap = new ClientBootstrap(CHANNEL_FACTORY); // config // @see org.jboss.netty.channel.socket.SocketChannelConfig bootstrap.setOption("keepAlive", true); bootstrap.setOption("tcpNoDelay", true); bootstrap.setOption("connectTimeoutMillis", getConnectTimeout()); final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this); ChannelPipeline pipeline = Channels.pipeline(); pipeline.addLast("decoder", adapter.getDecoder()); pipeline.addLast("encoder", adapter.getEncoder()); pipeline.addLast("handler", nettyHandler); return pipeline; } }); } 比较简单，主要就是通过 NettyHandler 进入数据处理。当然，编解码是少不了的。 @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { handler.received(channel, e.getMessage()); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } } @Override public void received(Channel channel, Object message) throws RemotingException { final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel); if (message instanceof Request) { // handle request. Request request = (Request) message; if (request.isEvent()) { 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) { 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); } } // org.apache.dubbo.remoting.transport.netty4.NettyClientHandler#channelRead @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler); handler.received(channel, msg); } // org.apache.dubbo.remoting.transport.AbstractPeer#received @Override public void received(Channel ch, Object msg) throws RemotingException { if (closed) { return; } handler.received(ch, msg); }
View Code
2.8. 放置响应结果
前面多次提到响应结束后，结果将会被放到合适的位置，我们就简单看下它到底是怎么放置的呢？其实就是 CompletableFuture 的complete方法。
// 主动置位结果 private void doReceived(Response res) { if (res == null) { throw new IllegalStateException("response cannot be null"); } if (res.getStatus() == Response.OK) { // 放置结果后结束 this.complete(res.getResult()); } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) { this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage())); } else { this.completeExceptionally(new RemotingException(channel, res.getErrorMessage())); } // the result is returning, but the caller thread may still waiting // to avoid endless waiting for whatever reason, notify caller thread to return. if (executor != null && executor instanceof ThreadlessExecutor) { ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor; if (threadlessExecutor.isWaiting()) { threadlessExecutor.notifyReturn(new IllegalStateException("The result has returned, but the biz thread is still waiting" + " which is not an expected state, interrupt the thread manually by returning an exception.")); } } }
可以看出，同步和异步调用的区别主要是线程池的处理，以级后续事件的触发时机不同。同步调用在框架层面的假设是，发送消息之后，很快就会进行get() 操作，所以此时只需将就绪事件放入队列即可。而异步调用则可能没有后续的用户驱动，所以不能有卡点的出现，所以直接运行相应的结果通知，将结果放置到正确的位置。至于客户端来取或不来取，整体都不景程。
超时机制，是通过一个定时器，到点检查，检查到即超时。如果结果先出来，那么，主动将定时器取消，一切正常。因为定时器是另外的线程池进行处理，不受当前处理线程的影响，所以可以很好地控制超时。不管是客户端超时，还是服务端超时，都一概处理。
最后，再说下超时时的消息描述信息，因为这可能给我排查问题带来极大的便利。
2.9. 超时异常信息解析
判断是客户端超时还是服务端超时，是通过是否将消息发送出去为准的。实际上，它并不能区分出到底是客户端发送得晚了，还是服务端真的处理慢了。也就是说，当客户端自己慢的时候，它很可能认为是服务端超时了。而且，客户端是假设服务端一发送加响应消息，客户端就立即能收到结果，然后就以当时时间来判定服务端的处理时间。然而这样的判断方式，在客户端自身压力很大的情况下，仍然是有失偏颇的。代码描述如下：
// 错误信息详细描述 // org.apache.dubbo.remoting.exchange.support.DefaultFuture#getTimeoutMessage private String getTimeoutMessage(boolean scan) { long nowTimestamp = System.currentTimeMillis(); return (sent > 0 ? "Waiting server-side response timeout" : "Sending request timeout in client-side") + (scan ? " by scan timer" : "") + ". start time: " + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(start))) + ", end time: " + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(nowTimestamp))) + "," + (sent > 0 ? " client elapsed: " + (sent - start) + " ms, server elapsed: " + (nowTimestamp - sent) : " elapsed: " + (nowTimestamp - start)) + " ms, timeout: " + timeout + " ms, request: " + (logger.isDebugEnabled() ? request : getRequestWithoutData()) + ", channel: " + channel.getLocalAddress() + " -> " + channel.getRemoteAddress(); }
3. server端超时实现
通过上一节，我们可以看到客户端的超时机制比较简单，但是实际上也是非常完善的。那么，对于服务端是否也有同样的一套东西呢？事实上，要控制服务端的超时，难度要比客户端大：一是因为服务端作为服务提供者，应该是要保证服务正常处理，而不是边处理边检查是否超时；二是服务端如果发现了超时，应该怎么对客户端说呢？抛出异常或者不返回消息？客户端因为是终端，他可以忽略结果即可，但服务端这样做却是不太合适的。另外，服务端计算超时的方式是不完善的，因为超时一般是针对客户端而言，因为整体链路除了服务端的处理时间，还有网络传输、处理时间，客户端自行处理的时间等等，所以服务端的超时标准不太可靠。
3.1. 业务处理线程池接入
当网络数据就绪之后，会将数据提交到业务线程池进行处理，也就是说io线程和业务线程是分离的。这是一般的处理方式，避免阻塞io线程，也方便扩展业务线程。我们理解，其实要做超时，在这个地方是比较合适的。
// 服务端消息接入 // org.apache.dubbo.remoting.transport.dispatcher.all.AllChannelHandler#received @Override public void received(Channel channel, Object message) throws RemotingException { ExecutorService executor = getPreferredExecutorService(message); try { // 交由对应的线程池异步处理, 状态为 RECEIVED // 此处其实可能存在阻塞等待问题 executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)); } catch (Throwable t) { if(message instanceof Request && t instanceof RejectedExecutionException){ sendFeedback(channel, (Request) message, t); return; } throw new ExecutionException(message, channel, getClass() + " error when process received event .", t); } } // org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeHandler#received @Override public void received(Channel channel, Object message) throws RemotingException { final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel); if (message instanceof Request) { // handle request. Request request = (Request) message; if (request.isEvent()) { handlerEvent(channel, request); } else { if (request.isTwoWay()) { // handleRequest handleRequest(exchangeChannel, request); } 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); } } // org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeHandler#handleRequest 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 { CompletionStage<Object> future = handler.reply(channel, msg); // 异步等待结果响应回调 future.whenComplete((appResult, t) -> { try { // 没有异常，就是正常 if (t == null) { res.setStatus(Response.OK); res.setResult(appResult); } else { res.setStatus(Response.SERVICE_ERROR); res.setErrorMessage(StringUtils.toString(t)); } channel.send(res); } catch (RemotingException e) { // 在客户端关闭连接时，发送消息将会失败 logger.warn("Send result to consumer failed, channel is " + channel + ", msg is " + e); } }); } catch (Throwable e) { res.setStatus(Response.SERVICE_ERROR); res.setErrorMessage(StringUtils.toString(e)); channel.send(res); } } // org.apache.dubbo.rpc.proxy.AbstractProxyInvoker#invoke @Override public Result invoke(Invocation invocation) throws RpcException { try { // 调用正常的rpc方法 Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()); CompletableFuture<Object> future = wrapWithFuture(value); CompletableFuture<AppResponse> appResponseFuture = future.handle((obj, t) -> { AppResponse result = new AppResponse(invocation); if (t != null) { if (t instanceof CompletionException) { result.setException(t.getCause()); } else { result.setException(t); } } else { result.setValue(obj); } return result; }); // 包装返回结果 return new AsyncRpcResult(appResponseFuture, invocation); } catch (InvocationTargetException e) { if (RpcContext.getContext().isAsyncStarted() && !RpcContext.getContext().stopAsync()) { logger.error("Provider async started, but got an exception from the original method, cannot write the exception back to consumer because an async result may have returned the new thread.", e); } return AsyncRpcResult.newDefaultAsyncResult(null, e.getTargetException(), invocation); } catch (Throwable e) { throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e); } }
server端确实也使用了一个独立的线程池来处理业务，但是并没有看到相应的外围超时处理。这是比较疑惑的，因为它已经错过了最佳判断超时的时机了。那么，是否服务端就不能提供超时功能了呢？
3.2. server解析timeout信息
server端仅在特殊情况下才会处理超时。它是在 TimeoutFilter 做的简单处理，仅将结果清空，然后正常返回了。
// org.apache.dubbo.rpc.protocol.FilterNode#invoke @Override public Result invoke(Invocation invocation) throws RpcException { Result asyncResult; try { asyncResult = filter.invoke(next, invocation); } catch (Exception e) { if (filter instanceof ListenableFilter) { ListenableFilter listenableFilter = ((ListenableFilter) filter); try { Filter.Listener listener = listenableFilter.listener(invocation); if (listener != null) { listener.onError(e, invoker, invocation); } } finally { listenableFilter.removeListener(invocation); } } else if (filter instanceof Filter.Listener) { Filter.Listener listener = (Filter.Listener) filter; listener.onError(e, invoker, invocation); } throw e; } finally { } return asyncResult.whenCompleteWithContext((r, t) -> { if (filter instanceof ListenableFilter) { ListenableFilter listenableFilter = ((ListenableFilter) filter); Filter.Listener listener = listenableFilter.listener(invocation); try { if (listener != null) { if (t == null) { listener.onResponse(r, invoker, invocation); } else { listener.onError(t, invoker, invocation); } } } finally { listenableFilter.removeListener(invocation); } } else if (filter instanceof Filter.Listener) { Filter.Listener listener = (Filter.Listener) filter; if (t == null) { listener.onResponse(r, invoker, invocation); } else { listener.onError(t, invoker, invocation); } } }); } // org.apache.dubbo.rpc.filter.ContextFilter#invoke @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { Map<String, Object> attachments = invocation.getObjectAttachments(); if (attachments != null) { Map<String, Object> newAttach = new HashMap<>(attachments.size()); for (Map.Entry<String, Object> entry : attachments.entrySet()) { String key = entry.getKey(); if (!UNLOADING_KEYS.contains(key)) { newAttach.put(key, entry.getValue()); } } attachments = newAttach; } RpcContext context = RpcContext.getContext(); context.setInvoker(invoker) .setInvocation(invocation) // .setAttachments(attachments) // merged from dubbox .setLocalAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort()); String remoteApplication = (String) invocation.getAttachment(REMOTE_APPLICATION_KEY); if (StringUtils.isNotEmpty(remoteApplication)) { context.setRemoteApplicationName(remoteApplication); } else { context.setRemoteApplicationName((String) context.getAttachment(REMOTE_APPLICATION_KEY)); } // 此处为服务端的超时实现，通过 _TO:xx 配置，由客户端传导到服务端进行控制，当超时时，结果将被清空 // 即此处的超时是伪超时，客户端实现的超时才是真实的 long timeout = RpcUtils.getTimeout(invocation, -1); if (timeout != -1) { context.set(TIME_COUNTDOWN_KEY, TimeoutCountDown.newCountDown(timeout, TimeUnit.MILLISECONDS)); } // merged from dubbox // we may already added some attachments into RpcContext before this filter (e.g. in rest protocol) if (attachments != null) { if (context.getObjectAttachments() != null) { context.getObjectAttachments().putAll(attachments); } else { context.setObjectAttachments(attachments); } } if (invocation instanceof RpcInvocation) { ((RpcInvocation) invocation).setInvoker(invoker); } try { context.clearAfterEachInvoke(false); return invoker.invoke(invocation); } finally { context.clearAfterEachInvoke(true); // IMPORTANT! For async scenario, we must remove context from current thread, so we always create a new RpcContext for the next invoke for the same thread. RpcContext.removeContext(true); RpcContext.removeServerContext(); } } // org.apache.dubbo.rpc.filter.TimeoutFilter#invoke @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { return invoker.invoke(invocation); } // TimeoutFilter @Override public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) { // "timeout-countdown" Object obj = RpcContext.getContext().get(TIME_COUNTDOWN_KEY); if (obj != null) { // 超时后，将结果清空 TimeoutCountDown countDown = (TimeoutCountDown) obj; if (countDown.isExpired()) { ((AppResponse) appResponse).clear(); // clear response in case of timeout. if (logger.isWarnEnabled()) { logger.warn("invoke timed out. method: " + invocation.getMethodName() + " arguments: " + Arrays.toString(invocation.getArguments()) + " , url is " + invoker.getUrl() + ", invoke elapsed " + countDown.elapsedMillis() + " ms."); } } } }
服务端的超时控制，并非像客户端那样，可以直接断开服务，或者丢弃连接。而是需要谨慎处理，此处为清空结果。这也许不是大家想要的超时。
3.3. 服务端server的开启过程
服务端作为提供者，会将自己所有的必要的服务注册到注册中心，所以在在启动时会使用netty服务框架，打开网络端口。这个过程是在导出第一个service的时候进行的。感兴趣的自行翻阅。
// 导出服务时会打开远程连接，对外提供端口服务 // org.apache.dubbo.config.ServiceConfig#doExportUrlsFor1Protocol private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs, int protocolConfigNum) { String name = protocolConfig.getName(); if (StringUtils.isEmpty(name)) { name = DUBBO; } ... Exporter<?> exporter = PROTOCOL.export(wrapperInvoker); exporters.add(exporter); ... this.urls.add(url); } // org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#export @Override public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { URL url = invoker.getUrl(); // export service. String key = serviceKey(url); DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap); exporterMap.put(key, exporter); //export an stub service for dispatching event Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT); Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false); if (isStubSupportEvent && !isCallbackservice) { String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY); if (stubServiceMethods == null || stubServiceMethods.length() == 0) { if (logger.isWarnEnabled()) { logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) + "], has set stubproxy support event ,but no stub methods founded.")); } } } openServer(url); optimizeSerialization(url); return exporter; } private void openServer(URL url) { // 一个ip:port, 对应一个server // find server. String key = url.getAddress(); //client can export a service which's only for server to invoke boolean isServer = url.getParameter(IS_SERVER_KEY, true); if (isServer) { ProtocolServer server = serverMap.get(key); if (server == null) { synchronized (this) { server = serverMap.get(key); if (server == null) { serverMap.put(key, createServer(url)); } } } else { // server supports reset, use together with override server.reset(url); } } } // org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#createServer private ProtocolServer createServer(URL url) { url = URLBuilder.from(url) // send readonly event when server closes, it's enabled by default .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()) // enable heartbeat by default .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT)) .addParameter(CODEC_KEY, DubboCodec.NAME) .build(); String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER); if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException("Unsupported server type: " + str + ", url: " + url); } ExchangeServer server; try { // 通过 header=org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger 进行查找合适的网络传输组件 server = Exchangers.bind(url, requestHandler); } catch (RemotingException e) { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } str = url.getParameter(CLIENT_KEY); if (str != null && str.length() > 0) { Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(); if (!supportedTypes.contains(str)) { throw new RpcException("Unsupported client type: " + str); } } return new DubboProtocolServer(server); } // 最终在 HeeaderExchanger 里面加载transporter // org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger#bind @Override public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); } // org.apache.dubbo.remoting.Transporters#bind(org.apache.dubbo.common.URL, org.apache.dubbo.remoting.ChannelHandler...) public static RemotingServer bind(URL url, ChannelHandler... handlers) throws RemotingException { if (url == null) { throw new IllegalArgumentException("url == null"); } if (handlers == null || handlers.length == 0) { throw new IllegalArgumentException("handlers == null"); } ChannelHandler handler; if (handlers.length == 1) { handler = handlers[0]; } else { handler = new ChannelHandlerDispatcher(handlers); } // 默认为取 netty 的配置 // @SPI("netty") // public interface Transporter // 而netty的配置是: netty4=org.apache.dubbo.remoting.transport.netty4.NettyTransporter // netty=org.apache.dubbo.remoting.transport.netty4.NettyTransporter return getTransporter().bind(url, handler); } // org.apache.dubbo.remoting.transport.netty4.NettyTransporter#bind @Override public RemotingServer bind(URL url, ChannelHandler handler) throws RemotingException { return new NettyServer(url, handler); } // org.apache.dubbo.remoting.transport.netty4.NettyServer#NettyServer public NettyServer(URL url, ChannelHandler handler) throws RemotingException { // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants. // the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler super(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), ChannelHandlers.wrap(handler, url)); } // org.apache.dubbo.remoting.transport.AbstractServer#AbstractServer public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); localAddress = getUrl().toInetSocketAddress(); String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost()); int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort()); if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { bindIp = ANYHOST_VALUE; } bindAddress = new InetSocketAddress(bindIp, bindPort); this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS); try { doOpen(); if (logger.isInfoEnabled()) { logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress()); } } catch (Throwable t) { throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t); } executor = executorRepository.createExecutorIfAbsent(url); } // org.apache.dubbo.remoting.transport.netty4.NettyServer#doOpen /** * Init and start netty server * * @throws Throwable */ @Override protected void doOpen() throws Throwable { bootstrap = new ServerBootstrap(); bossGroup = NettyEventLoopFactory.eventLoopGroup(1, "NettyServerBoss"); workerGroup = NettyEventLoopFactory.eventLoopGroup( getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), "NettyServerWorker"); // 服务端功能接入处理器 final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this); channels = nettyServerHandler.getChannels(); boolean keepalive = getUrl().getParameter(KEEP_ALIVE_KEY, Boolean.FALSE); bootstrap.group(bossGroup, workerGroup) .channel(NettyEventLoopFactory.serverSocketChannelClass()) .option(ChannelOption.SO_REUSEADDR, Boolean.TRUE) .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE) .childOption(ChannelOption.SO_KEEPALIVE, keepalive) .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // FIXME: should we use getTimeout()? int idleTimeout = UrlUtils.getIdleTimeout(getUrl()); NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); if (getUrl().getParameter(SSL_ENABLED_KEY, false)) { ch.pipeline().addLast("negotiation", SslHandlerInitializer.sslServerHandler(getUrl(), nettyServerHandler)); } // 编解码, 正式处理器 // pipeline, encoder -> handler 出站, decoder -> handler 入站 ch.pipeline() .addLast("decoder", adapter.getDecoder()) .addLast("encoder", adapter.getEncoder()) .addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS)) .addLast("handler", nettyServerHandler); } }); // bind ChannelFuture channelFuture = bootstrap.bind(getBindAddress()); channelFuture.syncUninterruptibly(); channel = channelFuture.channel(); } // 处理各种网络事件的分发 // org.apache.dubbo.remoting.transport.dispatcher.ChannelEventRunnable#run @Override public void run() { if (state == ChannelState.RECEIVED) { try { handler.received(channel, message); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel + ", message is " + message, e); } } else { switch (state) { case CONNECTED: try { handler.connected(channel); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case DISCONNECTED: try { handler.disconnected(channel); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case SENT: try { handler.sent(channel, message); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel + ", message is " + message, e); } break; case CAUGHT: try { handler.caught(channel, exception); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel + ", message is: " + message + ", exception is " + exception, e); } break; default: logger.warn("unknown state: " + state + ", message is " + message); } } } // org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#requestHandler private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() { @Override public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException { if (!(message instanceof Invocation)) { throw new RemotingException(channel, "Unsupported request: " + (message == null ? null : (message.getClass().getName() + ": " + message)) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress()); } Invocation inv = (Invocation) message; Invoker<?> invoker = getInvoker(channel, inv); // need to consider backward-compatibility if it's a callback if (Boolean.TRUE.toString().equals(inv.getObjectAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) { String methodsStr = invoker.getUrl().getParameters().get("methods"); boolean hasMethod = false; if (methodsStr == null || !methodsStr.contains(",")) { hasMethod = inv.getMethodName().equals(methodsStr); } else { String[] methods = methodsStr.split(","); for (String method : methods) { if (inv.getMethodName().equals(method)) { hasMethod = true; break; } } } if (!hasMethod) { logger.warn(new IllegalStateException("The methodName " + inv.getMethodName() + " not found in callback service interface ,invoke will be ignored." + " please update the api interface. url is:" + invoker.getUrl()) + " ,invocation is :" + inv); return null; } } RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress()); Result result = invoker.invoke(inv); return result.thenApply(Function.identity()); } ... // org.apache.dubbo.rpc.protocol.FilterNode#invoke @Override public Result invoke(Invocation invocation) throws RpcException { Result asyncResult; try { asyncResult = filter.invoke(next, invocation); } catch (Exception e) { if (filter instanceof ListenableFilter) { ListenableFilter listenableFilter = ((ListenableFilter) filter); try { Filter.Listener listener = listenableFilter.listener(invocation); if (listener != null) { listener.onError(e, invoker, invocation); } } finally { listenableFilter.removeListener(invocation); } } else if (filter instanceof Filter.Listener) { Filter.Listener listener = (Filter.Listener) filter; listener.onError(e, invoker, invocation); } throw e; } finally { } // 结果回调通知, 用于监控、超时处理 之类的扩展点 return asyncResult.whenCompleteWithContext((r, t) -> { if (filter instanceof ListenableFilter) { ListenableFilter listenableFilter = ((ListenableFilter) filter); Filter.Listener listener = listenableFilter.listener(invocation); try { if (listener != null) { if (t == null) { listener.onResponse(r, invoker, invocation); } else { listener.onError(t, invoker, invocation); } } } finally { listenableFilter.removeListener(invocation); } } else if (filter instanceof Filter.Listener) { Filter.Listener listener = (Filter.Listener) filter; if (t == null) { listener.onResponse(r, invoker, invocation); } else { listener.onError(t, invoker, invocation); } } }); } // org.apache.dubbo.rpc.proxy.AbstractProxyInvoker#invoke @Override public Result invoke(Invocation invocation) throws RpcException { try { // 通过代理，调用用户的rpc实现 // JavaassistProxyFactory Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()); // 使用future 封装返回 CompletableFuture<Object> future = wrapWithFuture(value); CompletableFuture<AppResponse> appResponseFuture = future.handle((obj, t) -> { AppResponse result = new AppResponse(invocation); if (t != null) { if (t instanceof CompletionException) { result.setException(t.getCause()); } else { result.setException(t); } } else { result.setValue(obj); } return result; }); return new AsyncRpcResult(appResponseFuture, invocation); } catch (InvocationTargetException e) { if (RpcContext.getContext().isAsyncStarted() && !RpcContext.getContext().stopAsync()) { logger.error("Provider async started, but got an exception from the original method, cannot write the exception back to consumer because an async result may have returned the new thread.", e); } return AsyncRpcResult.newDefaultAsyncResult(null, e.getTargetException(), invocation); } catch (Throwable e) { throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e); } }
View Code
虽然本节是讲server端的超时控制的，但是很明显这方便的讲述也很少，原因是本来就没打算在server端实现超时。我们要做的，也许只是验证一下而已。
4. 总结
dubbo中的超时设置，可以在服务端、消费端，而且官方建议是设置在服务端，客户端做特殊处理即可。原因是服务端更清楚接口的性能情况。这是完全理解的。但它会给人一种感觉，好像是真的服务端真的实现了超时处理。然而实际情况却是，它仅将该参数传导到客户端，然后由客户端来控制了。这倒是和直觉不太一样，但是谁能说他的直觉就是对的呢。
其实要想实现真正的服务端超时，也是可以的。同样，它也需要借助一些额外的线程池。比如，在接收其实要想实现真正的服务端超时，也是可以的。同样，它也需要借助一些额外的线程池。比如，在接收完数据之后，需要添加到业务线程池中进行处理，此时在提交之前写入一个开始时间，然后在线程池真正处理的时候，与当前时间运算，超时后就不再进行后续的计算逻辑，而是直接响应客户端超时。这个思路很简单，但至于为什么没有被实现，也许会有其他的考量，我们只讨论局部思路了。
注：本篇使用dubbo版本为 2.7.0
