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;
27.  
28.  
29. /**
30. *请求信息
31. *@since0.0.7
32. */
33. privatefinalConcurrentHashMapresponseMap;
34.  
35.  
36. /**
37. *新建
38. *@paramrequestMap请求Map
39. *@paramresponseMap结果map
40. *@since0.0.7
41. */
42. publicTimeoutCheckThread(ConcurrentHashMaprequestMap,
43. ConcurrentHashMapresponseMap){
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()){
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/