偷偷摘套内射激情视频,久久精品99国产国产精,中文字幕无线乱码人妻,中文在线中文a,性爽19p

Java 從零開始手寫 RPC-timeout 超時(shí)處理

作者：老馬嘯西風(fēng)
開發(fā) 后端
前面我們實(shí)現(xiàn)了通用的 rpc，但是存在一個(gè)問題，同步獲取響應(yīng)的時(shí)候沒有超時(shí)處理。如果 server 掛掉了，或者處理太慢，客戶端也不可能一直傻傻的等。

必要性

前面我們實(shí)現(xiàn)了通用的 rpc，但是存在一個(gè)問題，同步獲取響應(yīng)的時(shí)候沒有超時(shí)處理。

如果 server 掛掉了，或者處理太慢，客戶端也不可能一直傻傻的等。

當(dāng)外部的調(diào)用超過指定的時(shí)間后，就直接報(bào)錯(cuò)，避免無意義的資源消耗。

思路

調(diào)用的時(shí)候，將開始時(shí)間保留。

獲取的時(shí)候檢測是否超時(shí)。

同時(shí)創(chuàng)建一個(gè)線程，用來檢測是否有超時(shí)的請求。

實(shí)現(xiàn)

思路

調(diào)用的時(shí)候，將開始時(shí)間保留。

獲取的時(shí)候檢測是否超時(shí)。

同時(shí)創(chuàng)建一個(gè)線程，用來檢測是否有超時(shí)的請求。

超時(shí)檢測線程

為了不影響正常業(yè)務(wù)的性能，我們另起一個(gè)線程檢測調(diào)用是否已經(jīng)超時(shí)。

  1. package com.github.houbb.rpc.client.invoke.impl; 
  2.  
  3.  
  4. import com.github.houbb.heaven.util.common.ArgUtil; 
  5. import com.github.houbb.rpc.common.rpc.domain.RpcResponse; 
  6. import com.github.houbb.rpc.common.rpc.domain.impl.RpcResponseFactory; 
  7. import com.github.houbb.rpc.common.support.time.impl.Times; 
  8.  
  9.  
  10. import java.util.Map; 
  11. import java.util.concurrent.ConcurrentHashMap; 
  12.  
  13.  
  14. /** 
  15.  * 超時(shí)檢測線程 
  16.  * @author binbin.hou 
  17.  * @since 0.0.7 
  18.  */ 
  19. public class TimeoutCheckThread implements Runnable{ 
  20.  
  21.  
  22.     /** 
  23.      * 請求信息 
  24.      * @since 0.0.7 
  25.      */ 
  26.     private final ConcurrentHashMap<String, Long> requestMap; 
  27.  
  28.  
  29.     /** 
  30.      * 請求信息 
  31.      * @since 0.0.7 
  32.      */ 
  33.     private final ConcurrentHashMap<String, RpcResponse> responseMap; 
  34.  
  35.  
  36.     /** 
  37.      * 新建 
  38.      * @param requestMap  請求 Map 
  39.      * @param responseMap 結(jié)果 map 
  40.      * @since 0.0.7 
  41.      */ 
  42.     public TimeoutCheckThread(ConcurrentHashMap<String, Long> requestMap, 
  43.                               ConcurrentHashMap<String, RpcResponse> responseMap) { 
  44.         ArgUtil.notNull(requestMap, "requestMap"); 
  45.         this.requestMap = requestMap; 
  46.         this.responseMap = responseMap; 
  47.     } 
  48.  
  49.  
  50.     @Override 
  51.     public void run() { 
  52.         for(Map.Entry<String, Long> entry : requestMap.entrySet()) { 
  53.             long expireTime = entry.getValue(); 
  54.             long currentTime = Times.time(); 
  55.  
  56.  
  57.             if(currentTime > expireTime) { 
  58.                 final String key = entry.getKey(); 
  59.                 // 結(jié)果設(shè)置為超時(shí)，從請求 map 中移除 
  60.                 responseMap.putIfAbsent(key, RpcResponseFactory.timeout()); 
  61.                 requestMap.remove(key); 
  62.             } 
  63.         } 
  64.     } 
  65.  
  66.  
  67. }  

這里主要存儲(chǔ)請求，響應(yīng)的時(shí)間，如果超時(shí)，則移除對應(yīng)的請求。

線程啟動(dòng)

在 DefaultInvokeService 初始化時(shí)啟動(dòng)：

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

DefaultInvokeService

原來的設(shè)置結(jié)果，獲取結(jié)果是沒有考慮時(shí)間的，這里加一下對應(yīng)的判斷。

設(shè)置請求時(shí)間

•添加請求 addRequest

會(huì)將過時(shí)的時(shí)間直接放入 map 中。

因?yàn)榉湃胧且淮尾僮鳎樵兛赡苁嵌啻巍?/p>

所以時(shí)間在放入的時(shí)候計(jì)算完成。

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

設(shè)置請求結(jié)果

•添加響應(yīng) addResponse

1.如果 requestMap 中已經(jīng)不存在這個(gè)請求信息，則說明可能超時(shí)，直接忽略存入結(jié)果。

2.此時(shí)檢測是否出現(xiàn)超時(shí)，超時(shí)直接返回超時(shí)信息。

3.放入信息后，通知其他等待的所有進(jìn)程。

  1. @Override 
  2. public InvokeService addResponse(String seqId, RpcResponse rpcResponse) { 
  3.     // 1. 判斷是否有效 
  4.     Long expireTime = this.requestMap.get(seqId); 
  5.     // 如果為空，可能是這個(gè)結(jié)果已經(jīng)超時(shí)了，被定時(shí) job 移除之后，響應(yīng)結(jié)果才過來。直接忽略 
  6.     if(ObjectUtil.isNull(expireTime)) { 
  7.         return this; 
  8.     } 
  9.  
  10.  
  11.     //2. 判斷是否超時(shí) 
  12.     if(Times.time() > expireTime) { 
  13.         LOG.info("[Client] seqId:{} 信息已超時(shí)，直接返回超時(shí)結(jié)果。", seqId); 
  14.         rpcResponse = RpcResponseFactory.timeout(); 
  15.     } 
  16.  
  17.  
  18.     // 這里放入之前，可以添加判斷。 
  19.     // 如果 seqId 必須處理請求集合中，才允許放入?；蛘咧苯雍雎詠G棄。 
  20.     // 通知所有等待方 
  21.     responseMap.putIfAbsent(seqId, rpcResponse); 
  22.     LOG.info("[Client] 獲取結(jié)果信息，seqId: {}, rpcResponse: {}", seqId, rpcResponse); 
  23.     LOG.info("[Client] seqId:{} 信息已經(jīng)放入，通知所有等待方", seqId); 
  24.     // 移除對應(yīng)的 requestMap 
  25.     requestMap.remove(seqId); 
  26.     LOG.info("[Client] seqId:{} remove from request map", seqId); 
  27.     synchronized (this) { 
  28.         this.notifyAll(); 
  29.     } 
  30.     return this; 
  31. } 

獲取請求結(jié)果

•獲取相應(yīng) getResponse

1.如果結(jié)果存在，直接返回響應(yīng)結(jié)果

2.否則進(jìn)入等待。

3.等待結(jié)束后獲取結(jié)果。

  1. @Override 
  2. public RpcResponse getResponse(String seqId) { 
  3.     try { 
  4.         RpcResponse rpcResponse = this.responseMap.get(seqId); 
  5.         if(ObjectUtil.isNotNull(rpcResponse)) { 
  6.             LOG.info("[Client] seq {} 對應(yīng)結(jié)果已經(jīng)獲取: {}", seqId, rpcResponse); 
  7.             return rpcResponse; 
  8.         } 
  9.         // 進(jìn)入等待 
  10.         while (rpcResponse == null) { 
  11.             LOG.info("[Client] seq {} 對應(yīng)結(jié)果為空，進(jìn)入等待", seqId); 
  12.             // 同步等待鎖 
  13.             synchronized (this) { 
  14.                 this.wait(); 
  15.             } 
  16.             rpcResponse = this.responseMap.get(seqId); 
  17.             LOG.info("[Client] seq {} 對應(yīng)結(jié)果已經(jīng)獲取: {}", seqId, rpcResponse); 
  18.         } 
  19.         return rpcResponse; 
  20.     } catch (InterruptedException e) { 
  21.         throw new RpcRuntimeException(e); 
  22.     } 
  23. } 

可以發(fā)現(xiàn)獲取部分的邏輯沒變，因?yàn)槌瑫r(shí)會(huì)返回一個(gè)超時(shí)對象：RpcResponseFactory.timeout();

這是一個(gè)非常簡單的實(shí)現(xiàn)，如下：

  1. package com.github.houbb.rpc.common.rpc.domain.impl; 
  2.  
  3.  
  4. import com.github.houbb.rpc.common.exception.RpcTimeoutException; 
  5. import com.github.houbb.rpc.common.rpc.domain.RpcResponse; 
  6.  
  7.  
  8. /** 
  9.  * 響應(yīng)工廠類 
  10.  * @author binbin.hou 
  11.  * @since 0.0.7 
  12.  */ 
  13. public final class RpcResponseFactory { 
  14.  
  15.  
  16.     private RpcResponseFactory(){} 
  17.  
  18.  
  19.     /** 
  20.      * 超時(shí)異常信息 
  21.      * @since 0.0.7 
  22.      */ 
  23.     private static final DefaultRpcResponse TIMEOUT; 
  24.  
  25.  
  26.     static { 
  27.         TIMEOUT = new DefaultRpcResponse(); 
  28.         TIMEOUT.error(new RpcTimeoutException()); 
  29.     } 
  30.  
  31.  
  32.     /** 
  33.      * 獲取超時(shí)響應(yīng)結(jié)果 
  34.      * @return 響應(yīng)結(jié)果 
  35.      * @since 0.0.7 
  36.      */ 
  37.     public static RpcResponse timeout() { 
  38.         return TIMEOUT; 
  39.     } 
  40.  
  41.  
  42. } 

 響應(yīng)結(jié)果指定一個(gè)超時(shí)異常，這個(gè)異常會(huì)在代理處理結(jié)果時(shí)拋出：

  1. RpcResponse rpcResponse = proxyContext.invokeService().getResponse(seqId); 
  2. Throwable error = rpcResponse.error(); 
  3. if(ObjectUtil.isNotNull(error)) { 
  4.     throw error; 
  5. } 
  6. return rpcResponse.result(); 

測試代碼

服務(wù)端

我們故意把服務(wù)端的實(shí)現(xiàn)添加沉睡，其他保持不變。

  1. public class CalculatorServiceImpl implements CalculatorService { 
  2.  
  3.  
  4.     public CalculateResponse sum(CalculateRequest request) { 
  5.         int sum = request.getOne()+request.getTwo(); 
  6.  
  7.  
  8.         // 故意沉睡 3s 
  9.         try { 
  10.             TimeUnit.SECONDS.sleep(3); 
  11.         } catch (InterruptedException e) { 
  12.             e.printStackTrace(); 
  13.         } 
  14.  
  15.  
  16.         return new CalculateResponse(true, sum); 
  17.     } 
  18.  
  19.  
  20. } 

客戶端

設(shè)置對應(yīng)的超時(shí)時(shí)間為 1S，其他不變：

  1. public static void main(String[] args) { 
  2.     // 服務(wù)配置信息 
  3.     ReferenceConfig<CalculatorService> config = new DefaultReferenceConfig<CalculatorService>(); 
  4.     config.serviceId(ServiceIdConst.CALC); 
  5.     config.serviceInterface(CalculatorService.class); 
  6.     config.addresses("localhost:9527"); 
  7.     // 設(shè)置超時(shí)時(shí)間為1S 
  8.     config.timeout(1000); 
  9.  
  10.  
  11.     CalculatorService calculatorService = config.reference(); 
  12.     CalculateRequest request = new CalculateRequest(); 
  13.     request.setOne(10); 
  14.     request.setTwo(20); 
  15.  
  16.  
  17.     CalculateResponse response = calculatorService.sum(request); 
  18.     System.out.println(response); 
  19. } 

 日志如下：

  1. .log.integration.adaptors.stdout.StdOutExImpl' adapter. 
  2. [INFO] [2021-10-05 14:59:40.974] [main] [c.g.h.r.c.c.RpcClient.connect] - RPC 服務(wù)開始啟動(dòng)客戶端 
  3. ... 
  4. [INFO] [2021-10-05 14:59:42.504] [main] [c.g.h.r.c.c.RpcClient.connect] - RPC 服務(wù)啟動(dòng)客戶端完成，監(jiān)聽地址 localhost:9527 
  5. [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}]} 
  6. [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 
  7. [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 
  8. ... 
  9. Exception in thread "main" com.github.houbb.rpc.common.exception.RpcTimeoutException 
  10.     at com.github.houbb.rpc.common.rpc.domain.impl.RpcResponseFactory.<clinit>(RpcResponseFactory.java:23) 
  11.     at com.github.houbb.rpc.client.invoke.impl.DefaultInvokeService.addResponse(DefaultInvokeService.java:72) 
  12.     at com.github.houbb.rpc.client.handler.RpcClientHandler.channelRead0(RpcClientHandler.java:43) 
  13.     at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105) 
  14.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) 
  15.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) 
  16.     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) 
  17.     at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:241) 
  18.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) 
  19.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) 
  20.     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) 
  21.     at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310) 
  22.     at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284) 
  23.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) 
  24.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) 
  25.     at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) 
  26.     at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359) 
  27.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) 
  28.     at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) 
  29.     at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935) 
  30.     at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:138) 
  31.     at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645) 
  32.     at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580) 
  33.     at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497) 
  34.     at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459) 
  35.     at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858) 
  36.     at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138) 
  37.     at java.lang.Thread.run(Thread.java:748) 
  38. ... 
  39. [INFO] [2021-10-05 14:59:45.615] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:62e126d9a0334399904509acf8dfe0bb 信息已超時(shí)，直接返回超時(shí)結(jié)果。 
  40. [INFO] [2021-10-05 14:59:45.617] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] 獲取結(jié)果信息，seqId: 62e126d9a0334399904509acf8dfe0bb, rpcResponse: DefaultRpcResponse{seqId='null', error=com.github.houbb.rpc.common.exception.RpcTimeoutException, result=null} 
  41. [INFO] [2021-10-05 14:59:45.617] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:62e126d9a0334399904509acf8dfe0bb 信息已經(jīng)放入，通知所有等待方 
  42. [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 
  43. [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}} 
  44. [INFO] [2021-10-05 14:59:45.619] [main] [c.g.h.r.c.i.i.DefaultInvokeService.getResponse] - [Client] seq 62e126d9a0334399904509acf8dfe0bb 對應(yīng)結(jié)果已經(jīng)獲取: DefaultRpcResponse{seqId='null', error=com.github.houbb.rpc.common.exception.RpcTimeoutException, result=null} 
  45. ... 

可以發(fā)現(xiàn)，超時(shí)異常。

不足之處

對于超時(shí)的處理可以拓展為雙向的，比如服務(wù)端也可以指定超時(shí)限制，避免資源的浪費(fèi)。

 

責(zé)任編輯：姜華 來源： 今日頭條
Java timeout Java 基礎(chǔ)
相關(guān)推薦

2021-10-20 08:05:18

Java 序列化 Java 基礎(chǔ)

2021-10-13 08:21:52

Java websocket Java 基礎(chǔ)

2021-10-19 08:58:48

Java 語言 Java 基礎(chǔ)

2021-10-21 08:21:10

Java Reflect Java 基礎(chǔ)

2019-09-23 19:30:27

reduxreact.js前端

2021-10-14 08:39:17

Java Netty Java 基礎(chǔ)

2015-11-17 16:11:07

Code Review

2019-01-18 12:39:45

云計(jì)算PaaS公有云

2018-04-18 07:01:59

Docker容器虛擬機(jī)

2024-12-06 17:02:26

2020-07-02 15:32:23

Kubernetes容器架構(gòu)

2024-09-18 08:10:06

2024-10-05 00:00:06

HTTP請求處理容器

2010-05-26 17:35:08

配置Xcode SVN

2018-09-14 17:16:22

云計(jì)算軟件計(jì)算機(jī)網(wǎng)絡(luò)

2024-05-15 14:29:45

2021-10-27 08:10:15

Java 客戶端 Java 基礎(chǔ)

2015-05-06 09:36:05

Java語言從零開始學(xué)習(xí)

2015-10-15 14:16:24

2024-04-10 07:48:41

搜索引擎場景
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)