Java 从零开始手写 RPC-timeout 超时处理

必要性

前面我们实现了通用的从零超时处理 rpc，但是开始存在一个问题，同步获取响应的手写时候没有超时处理。

如果 server 挂掉了，从零超时处理或者处理太慢，开始客户端也不可能一直傻傻的手写等。

当外部的从零超时处理调用超过指定的时间后，就直接报错，开始避免无意义的手写资源消耗。

思路

调用的从零超时处理时候，将开始时间保留。开始

获取的手写时候检测是否超时。

同时创建一个线程，从零超时处理用来检测是开始否有超时的请求。

实现

思路

调用的手写时候，将开始时间保留。

获取的时候检测是否超时。

同时创建一个线程，用来检测是高防服务器否有超时的请求。

超时检测线程

为了不影响正常业务的性能，我们另起一个线程检测调用是否已经超时。

package com.github.houbb.rpc.client.invoke.impl; import com.github.houbb.heaven.util.common.ArgUtil; import com.github.houbb.rpc.common.rpc.domain.RpcResponse; import com.github.houbb.rpc.common.rpc.domain.impl.RpcResponseFactory; import com.github.houbb.rpc.common.support.time.impl.Times; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; /**  * 超时检测线程  * @author binbin.hou  * @since 0.0.7  */ public class TimeoutCheckThread implements Runnable{      /**      * 请求信息      * @since 0.0.7      */     private final ConcurrentHashMap<String, Long> requestMap;     /**      * 请求信息      * @since 0.0.7      */     private final ConcurrentHashMap<String, RpcResponse> responseMap;     /**      * 新建      * @param requestMap  请求 Map      * @param responseMap 结果 map      * @since 0.0.7      */     public TimeoutCheckThread(ConcurrentHashMap<String, Long> requestMap,                               ConcurrentHashMap<String, RpcResponse> responseMap) {          ArgUtil.notNull(requestMap, "requestMap");         this.requestMap = requestMap;         this.responseMap = responseMap;     }     @Override     public void run() {          for(Map.Entry<String, Long> entry : requestMap.entrySet()) {              long expireTime = entry.getValue();             long currentTime = Times.time();             if(currentTime > expireTime) {                  final String key = entry.getKey();                 // 结果设置为超时，从请求 map 中移除                 responseMap.putIfAbsent(key, RpcResponseFactory.timeout());                 requestMap.remove(key);             }         }     } }  

这里主要存储请求，响应的时间，如果超时，则移除对应的请求。

线程启动

在 DefaultInvokeService 初始化时启动：

final Runnable timeoutThread = new TimeoutCheckThread(requestMap, responseMap); Executors.newScheduledThreadPool(1)                 .scheduleAtFixedRate(timeoutThread,60, 60, TimeUnit.SECONDS); 

DefaultInvokeService

原来的设置结果，获取结果是没有考虑时间的，这里加一下对应的判断。

设置请求时间

•添加请求 addRequest

会将过时的时间直接放入 map 中。

因为放入是一次操作，查询可能是多次。

所以时间在放入的时候计算完成。

@Override public InvokeService addRequest(String seqId, long timeoutMills) {      LOG.info("[Client] start add request for seqId: { }, timeoutMills: { }", seqId,             timeoutMills);     final long expireTime = Times.time()+timeoutMills;     requestMap.putIfAbsent(seqId, expireTime);     return this; } 

设置请求结果

•添加响应 addResponse

1.如果 requestMap 中已经不存在这个请求信息，则说明可能超时，直接忽略存入结果。

2.此时检测是否出现超时，云服务器超时直接返回超时信息。

3.放入信息后，通知其他等待的所有进程。

@Override public InvokeService addResponse(String seqId, RpcResponse rpcResponse) {      // 1. 判断是否有效     Long expireTime = this.requestMap.get(seqId);     // 如果为空，可能是这个结果已经超时了，被定时 job 移除之后，响应结果才过来。直接忽略     if(ObjectUtil.isNull(expireTime)) {          return this;     }     //2. 判断是否超时     if(Times.time() > expireTime) {          LOG.info("[Client] seqId:{ } 信息已超时，直接返回超时结果。", seqId);         rpcResponse = RpcResponseFactory.timeout();     }     // 这里放入之前，可以添加判断。     // 如果 seqId 必须处理请求集合中，才允许放入。或者直接忽略丢弃。     // 通知所有等待方     responseMap.putIfAbsent(seqId, rpcResponse);     LOG.info("[Client] 获取结果信息，seqId: { }, rpcResponse: { }", seqId, rpcResponse);     LOG.info("[Client] seqId:{ } 信息已经放入，通知所有等待方", seqId);     // 移除对应的 requestMap     requestMap.remove(seqId);     LOG.info("[Client] seqId:{ } remove from request map", seqId);     synchronized (this) {          this.notifyAll();     }     return this; } 

获取请求结果

•获取相应 getResponse

1.如果结果存在，直接返回响应结果

2.否则进入等待。

3.等待结束后获取结果。

@Override public RpcResponse getResponse(String seqId) {      try {          RpcResponse rpcResponse = this.responseMap.get(seqId);         if(ObjectUtil.isNotNull(rpcResponse)) {              LOG.info("[Client] seq { } 对应结果已经获取: { }", seqId, rpcResponse);             return rpcResponse;         }         // 进入等待         while (rpcResponse == null) {              LOG.info("[Client] seq { } 对应结果为空，进入等待", seqId);             // 同步等待锁             synchronized (this) {                  this.wait();             }             rpcResponse = this.responseMap.get(seqId);             LOG.info("[Client] seq { } 对应结果已经获取: { }", seqId, rpcResponse);         }         return rpcResponse;     } catch (InterruptedException e) {          throw new RpcRuntimeException(e);     } } 

可以发现获取部分的逻辑没变，因为超时会返回一个超时对象：RpcResponseFactory.timeout();

这是一个非常简单的实现，如下：

package com.github.houbb.rpc.common.rpc.domain.impl; import com.github.houbb.rpc.common.exception.RpcTimeoutException; import com.github.houbb.rpc.common.rpc.domain.RpcResponse; /**  * 响应工厂类  * @author binbin.hou  * @since 0.0.7  */ public final class RpcResponseFactory {      private RpcResponseFactory(){ }     /**      * 超时异常信息      * @since 0.0.7      */     private static final DefaultRpcResponse TIMEOUT;     static {          TIMEOUT = new DefaultRpcResponse();         TIMEOUT.error(new RpcTimeoutException());     }     /**      * 获取超时响应结果      * @return 响应结果      * @since 0.0.7      */     public static RpcResponse timeout() {          return TIMEOUT;     } } 

 响应结果指定一个超时异常，这个异常会在代理处理结果时抛出：

RpcResponse rpcResponse = proxyContext.invokeService().getResponse(seqId); Throwable error = rpcResponse.error(); if(ObjectUtil.isNotNull(error)) {      throw error; } return rpcResponse.result(); 

测试代码

服务端

我们故意把服务端的实现添加沉睡，香港云服务器其他保持不变。

public class CalculatorServiceImpl implements CalculatorService {      public CalculateResponse sum(CalculateRequest request) {          int sum = request.getOne()+request.getTwo();         // 故意沉睡 3s         try {              TimeUnit.SECONDS.sleep(3);         } catch (InterruptedException e) {              e.printStackTrace();         }         return new CalculateResponse(true, sum);     } } 

客户端

设置对应的超时时间为 1S，其他不变：

public static void main(String[] args) {      // 服务配置信息     ReferenceConfig<CalculatorService> config = new DefaultReferenceConfig<CalculatorService>();     config.serviceId(ServiceIdConst.CALC);     config.serviceInterface(CalculatorService.class);     config.addresses("localhost:9527");     // 设置超时时间为1S     config.timeout(1000);     CalculatorService calculatorService = config.reference();     CalculateRequest request = new CalculateRequest();     request.setOne(10);     request.setTwo(20);     CalculateResponse response = calculatorService.sum(request);     System.out.println(response); } 

 日志如下：

.log.integration.adaptors.stdout.StdOutExImpl adapter. [INFO] [2021-10-05 14:59:40.974] [main] [c.g.h.r.c.c.RpcClient.connect] - RPC 服务开始启动客户端 ... [INFO] [2021-10-05 14:59:42.504] [main] [c.g.h.r.c.c.RpcClient.connect] - RPC 服务启动客户端完成，监听地址 localhost:9527 [INFO] [2021-10-05 14:59:42.533] [main] [c.g.h.r.c.p.ReferenceProxy.invoke] - [Client] start call remote with request: DefaultRpcRequest{ seqId=62e126d9a0334399904509acf8dfe0bb, createTime=1633417182525, serviceId=calc, methodName=sum, paramTypeNames=[com.github.houbb.rpc.server.facade.model.CalculateRequest], paramValues=[CalculateRequest{ one=10, two=20}]} [INFO] [2021-10-05 14:59:42.534] [main] [c.g.h.r.c.i.i.DefaultInvokeService.addRequest] - [Client] start add request for seqId: 62e126d9a0334399904509acf8dfe0bb, timeoutMills: 1000 [INFO] [2021-10-05 14:59:42.535] [main] [c.g.h.r.c.p.ReferenceProxy.invoke] - [Client] start call channel id: 00e04cfffe360988-000004bc-00000000-1178e1265e903c4c-7975626f ... Exception in thread "main" com.github.houbb.rpc.common.exception.RpcTimeoutException     at com.github.houbb.rpc.common.rpc.domain.impl.RpcResponseFactory.<clinit>(RpcResponseFactory.java:23)     at com.github.houbb.rpc.client.invoke.impl.DefaultInvokeService.addResponse(DefaultInvokeService.java:72)     at com.github.houbb.rpc.client.handler.RpcClientHandler.channelRead0(RpcClientHandler.java:43)     at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)     at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:241)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)     at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)     at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)     at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)     at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)     at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:138)     at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)     at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)     at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)     at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)     at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)     at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)     at java.lang.Thread.run(Thread.java:748) ... [INFO] [2021-10-05 14:59:45.615] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:62e126d9a0334399904509acf8dfe0bb 信息已超时，直接返回超时结果。 [INFO] [2021-10-05 14:59:45.617] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] 获取结果信息，seqId: 62e126d9a0334399904509acf8dfe0bb, rpcResponse: DefaultRpcResponse{ seqId=null, error=com.github.houbb.rpc.common.exception.RpcTimeoutException, result=null} [INFO] [2021-10-05 14:59:45.617] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:62e126d9a0334399904509acf8dfe0bb 信息已经放入，通知所有等待方 [INFO] [2021-10-05 14:59:45.618] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:62e126d9a0334399904509acf8dfe0bb remove from request map [INFO] [2021-10-05 14:59:45.618] [nioEventLoopGroup-2-1] [c.g.h.r.c.c.RpcClient.channelRead0] - [Client] response is :DefaultRpcResponse{ seqId=62e126d9a0334399904509acf8dfe0bb, error=null, result=CalculateResponse{ success=true, sum=30}} [INFO] [2021-10-05 14:59:45.619] [main] [c.g.h.r.c.i.i.DefaultInvokeService.getResponse] - [Client] seq 62e126d9a0334399904509acf8dfe0bb 对应结果已经获取: DefaultRpcResponse{ seqId=null, error=com.github.houbb.rpc.common.exception.RpcTimeoutException, result=null} ... 

可以发现，超时异常。

不足之处

对于超时的处理可以拓展为双向的，比如服务端也可以指定超时限制，避免资源的浪费。