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

时间：2022-09-09 19:18:26

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

必要性

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

如果 server 挂掉了，或者处理太慢，客户端也不可能一直傻傻的等。

当外部的调用超过指定的时间后，就直接报错，避免无意义的资源消耗。

思路

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

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

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

实现

思路

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

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

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

超时检测线程

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

  1. packagecom.github.houbb.rpc.client.invoke.impl;
  2.  
  3.  
  4. importcom.github.houbb.heaven.util.common.ArgUtil;
  5. importcom.github.houbb.rpc.common.rpc.domain.RpcResponse;
  6. importcom.github.houbb.rpc.common.rpc.domain.impl.RpcResponseFactory;
  7. importcom.github.houbb.rpc.common.support.time.impl.Times;
  8.  
  9.  
  10. importjava.util.Map;
  11. importjava.util.concurrent.ConcurrentHashMap;
  12.  
  13.  
  14. /**
  15. *超时检测线程
  16. *@authorbinbin.hou
  17. *@since0.0.7
  18. */
  19. publicclassTimeoutCheckThreadimplementsRunnable{
  20.  
  21.  
  22. /**
  23. *请求信息
  24. *@since0.0.7
  25. */
  26. privatefinalConcurrentHashMaprequestMap;,long>
  27.  
  28.  
  29. /**
  30. *请求信息
  31. *@since0.0.7
  32. */
  33. privatefinalConcurrentHashMapresponseMap;,rpcresponse>
  34.  
  35.  
  36. /**
  37. *新建
  38. *@paramrequestMap请求Map
  39. *@paramresponseMap结果map
  40. *@since0.0.7
  41. */
  42. publicTimeoutCheckThread(ConcurrentHashMaprequestMap,,long>
  43. ConcurrentHashMapresponseMap){,rpcresponse>
  44. ArgUtil.notNull(requestMap,"requestMap");
  45. this.requestMap=requestMap;
  46. this.responseMap=responseMap;
  47. }
  48.  
  49.  
  50. @Override
  51. publicvoidrun(){
  52. for(Map.Entryentry:requestMap.entrySet()){,long>
  53. longexpireTime=entry.getValue();
  54. longcurrentTime=Times.time();
  55.  
  56.  
  57. if(currentTime>expireTime){
  58. finalStringkey=entry.getKey();
  59. //结果设置为超时，从请求map中移除
  60. responseMap.putIfAbsent(key,RpcResponseFactory.timeout());
  61. requestMap.remove(key);
  62. }
  63. }
  64. }
  65.  
  66.  
  67. }

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

线程启动

在 DefaultInvokeService 初始化时启动：

  1. finalRunnabletimeoutThread=newTimeoutCheckThread(requestMap,responseMap);
  2. Executors.newScheduledThreadPool(1)
  3. .scheduleAtFixedRate(timeoutThread,60,60,TimeUnit.SECONDS);

DefaultInvokeService

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

设置请求时间

•添加请求 addRequest

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

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

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

  1. @Override
  2. publicInvokeServiceaddRequest(StringseqId,longtimeoutMills){
  3. LOG.info("[Client]startaddrequestforseqId:{},timeoutMills:{}",seqId,
  4. timeoutMills);
  5. finallongexpireTime=Times.time()+timeoutMills;
  6. requestMap.putIfAbsent(seqId,expireTime);
  7. returnthis;
  8. }

设置请求结果

•添加响应 addResponse

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

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

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

  1. @Override
  2. publicInvokeServiceaddResponse(StringseqId,RpcResponserpcResponse){
  3. //1.判断是否有效
  4. LongexpireTime=this.requestMap.get(seqId);
  5. //如果为空，可能是这个结果已经超时了，被定时job移除之后，响应结果才过来。直接忽略
  6. if(ObjectUtil.isNull(expireTime)){
  7. returnthis;
  8. }
  9.  
  10.  
  11. //2.判断是否超时
  12. if(Times.time()>expireTime){
  13. LOG.info("[Client]seqId:{}信息已超时，直接返回超时结果。",seqId);
  14. rpcResponse=RpcResponseFactory.timeout();
  15. }
  16.  
  17.  
  18. //这里放入之前，可以添加判断。
  19. //如果seqId必须处理请求集合中，才允许放入。或者直接忽略丢弃。
  20. //通知所有等待方
  21. responseMap.putIfAbsent(seqId,rpcResponse);
  22. LOG.info("[Client]获取结果信息，seqId:{},rpcResponse:{}",seqId,rpcResponse);
  23. LOG.info("[Client]seqId:{}信息已经放入，通知所有等待方",seqId);
  24. //移除对应的requestMap
  25. requestMap.remove(seqId);
  26. LOG.info("[Client]seqId:{}removefromrequestmap",seqId);
  27. synchronized(this){
  28. this.notifyAll();
  29. }
  30. returnthis;
  31. }

获取请求结果

•获取相应 getResponse

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

2.否则进入等待。

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

  1. @Override
  2. publicRpcResponsegetResponse(StringseqId){
  3. try{
  4. RpcResponserpcResponse=this.responseMap.get(seqId);
  5. if(ObjectUtil.isNotNull(rpcResponse)){
  6. LOG.info("[Client]seq{}对应结果已经获取:{}",seqId,rpcResponse);
  7. returnrpcResponse;
  8. }
  9. //进入等待
  10. while(rpcResponse==null){
  11. LOG.info("[Client]seq{}对应结果为空，进入等待",seqId);
  12. //同步等待锁
  13. synchronized(this){
  14. this.wait();
  15. }
  16. rpcResponse=this.responseMap.get(seqId);
  17. LOG.info("[Client]seq{}对应结果已经获取:{}",seqId,rpcResponse);
  18. }
  19. returnrpcResponse;
  20. }catch(InterruptedExceptione){
  21. thrownewRpcRuntimeException(e);
  22. }
  23. }

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

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

  1. packagecom.github.houbb.rpc.common.rpc.domain.impl;
  2.  
  3.  
  4. importcom.github.houbb.rpc.common.exception.RpcTimeoutException;
  5. importcom.github.houbb.rpc.common.rpc.domain.RpcResponse;
  6.  
  7.  
  8. /**
  9. *响应工厂类
  10. *@authorbinbin.hou
  11. *@since0.0.7
  12. */
  13. publicfinalclassRpcResponseFactory{
  14.  
  15.  
  16. privateRpcResponseFactory(){}
  17.  
  18.  
  19. /**
  20. *超时异常信息
  21. *@since0.0.7
  22. */
  23. privatestaticfinalDefaultRpcResponseTIMEOUT;
  24.  
  25.  
  26. static{
  27. TIMEOUT=newDefaultRpcResponse();
  28. TIMEOUT.error(newRpcTimeoutException());
  29. }
  30.  
  31.  
  32. /**
  33. *获取超时响应结果
  34. *@return响应结果
  35. *@since0.0.7
  36. */
  37. publicstaticRpcResponsetimeout(){
  38. returnTIMEOUT;
  39. }
  40.  
  41.  
  42. }

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

  1. RpcResponserpcResponse=proxyContext.invokeService().getResponse(seqId);
  2. Throwableerror=rpcResponse.error();
  3. if(ObjectUtil.isNotNull(error)){
  4. throwerror;
  5. }
  6. returnrpcResponse.result();

测试代码

服务端

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

  1. publicclassCalculatorServiceImplimplementsCalculatorService{
  2.  
  3.  
  4. publicCalculateResponsesum(CalculateRequestrequest){
  5. intsum=request.getOne()+request.getTwo();
  6.  
  7.  
  8. //故意沉睡3s
  9. try{
  10. TimeUnit.SECONDS.sleep(3);
  11. }catch(InterruptedExceptione){
  12. e.printStackTrace();
  13. }
  14.  
  15.  
  16. returnnewCalculateResponse(true,sum);
  17. }
  18.  
  19.  
  20. }

客户端

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

  1. publicstaticvoidmain(String[]args){
  2. //服务配置信息
  3. ReferenceConfigconfig=newDefaultReferenceConfig();
  4. config.serviceId(ServiceIdConst.CALC);
  5. config.serviceInterface(CalculatorService.class);
  6. config.addresses("localhost:9527");
  7. //设置超时时间为1S
  8. config.timeout(1000);
  9.  
  10.  
  11. CalculatorServicecalculatorService=config.reference();
  12. CalculateRequestrequest=newCalculateRequest();
  13. request.setOne(10);
  14. request.setTwo(20);
  15.  
  16.  
  17. CalculateResponseresponse=calculatorService.sum(request);
  18. System.out.println(response);
  19. }

日志如下：

  1. .log.integration.adaptors.stdout.StdOutExImpl'adapter.
  2. [INFO][2021-10-0514:59:40.974][main][c.g.h.r.c.c.RpcClient.connect]-RPC服务开始启动客户端
  3. ...
  4. [INFO][2021-10-0514:59:42.504][main][c.g.h.r.c.c.RpcClient.connect]-RPC服务启动客户端完成，监听地址localhost:9527
  5. [INFO][2021-10-0514:59:42.533][main][c.g.h.r.c.p.ReferenceProxy.invoke]-[Client]startcallremotewithrequest:DefaultRpcRequest{seqId='62e126d9a0334399904509acf8dfe0bb',createTime=1633417182525,serviceId='calc',methodName='sum',paramTypeNames=[com.github.houbb.rpc.server.facade.model.CalculateRequest],paramValues=[CalculateRequest{one=10,two=20}]}
  6. [INFO][2021-10-0514:59:42.534][main][c.g.h.r.c.i.i.DefaultInvokeService.addRequest]-[Client]startaddrequestforseqId:62e126d9a0334399904509acf8dfe0bb,timeoutMills:1000
  7. [INFO][2021-10-0514:59:42.535][main][c.g.h.r.c.p.ReferenceProxy.invoke]-[Client]startcallchannelid:00e04cfffe360988-000004bc-00000000-1178e1265e903c4c-7975626f
  8. ...
  9. Exceptioninthread"main"com.github.houbb.rpc.common.exception.RpcTimeoutException
  10. atcom.github.houbb.rpc.common.rpc.domain.impl.RpcResponseFactory.(RpcResponseFactory.java:23)
  11. atcom.github.houbb.rpc.client.invoke.impl.DefaultInvokeService.addResponse(DefaultInvokeService.java:72)
  12. atcom.github.houbb.rpc.client.handler.RpcClientHandler.channelRead0(RpcClientHandler.java:43)
  13. atio.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
  14. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
  15. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
  16. atio.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
  17. atio.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:241)
  18. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
  19. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
  20. atio.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
  21. atio.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
  22. atio.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)
  23. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
  24. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
  25. atio.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
  26. atio.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
  27. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
  28. atio.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
  29. atio.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
  30. atio.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:138)
  31. atio.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
  32. atio.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
  33. atio.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
  34. atio.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
  35. atio.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
  36. atio.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
  37. atjava.lang.Thread.run(Thread.java:748)
  38. ...
  39. [INFO][2021-10-0514:59:45.615][nioEventLoopGroup-2-1][c.g.h.r.c.i.i.DefaultInvokeService.addResponse]-[Client]seqId:62e126d9a0334399904509acf8dfe0bb信息已超时，直接返回超时结果。
  40. [INFO][2021-10-0514: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}
  41. [INFO][2021-10-0514:59:45.617][nioEventLoopGroup-2-1][c.g.h.r.c.i.i.DefaultInvokeService.addResponse]-[Client]seqId:62e126d9a0334399904509acf8dfe0bb信息已经放入，通知所有等待方
  42. [INFO][2021-10-0514:59:45.618][nioEventLoopGroup-2-1][c.g.h.r.c.i.i.DefaultInvokeService.addResponse]-[Client]seqId:62e126d9a0334399904509acf8dfe0bbremovefromrequestmap
  43. [INFO][2021-10-0514:59:45.618][nioEventLoopGroup-2-1][c.g.h.r.c.c.RpcClient.channelRead0]-[Client]responseis:DefaultRpcResponse{seqId='62e126d9a0334399904509acf8dfe0bb',error=null,result=CalculateResponse{success=true,sum=30}}
  44. [INFO][2021-10-0514:59:45.619][main][c.g.h.r.c.i.i.DefaultInvokeService.getResponse]-[Client]seq62e126d9a0334399904509acf8dfe0bb对应结果已经获取:DefaultRpcResponse{seqId='null',error=com.github.houbb.rpc.common.exception.RpcTimeoutException,result=null}
  45. ...

可以发现，超时异常。

不足之处

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

原文链接：https://www.toutiao.com/a7018512258305278500/

标签：

相关文章