Java 從零開始手寫 RPC-timeout 超時處理

作者：老馬嘯西風(fēng) 2021-10-29 08:07:30

開發(fā) 后端

前面我們實現(xiàn)了通用的 rpc，但是存在一個問題，同步獲取響應(yīng)的時候沒有超時處理。如果 server 掛掉了，或者處理太慢，客戶端也不可能一直傻傻的等。

必要性

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

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

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

思路

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

獲取的時候檢測是否超時。

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

實現(xiàn)

思路

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

獲取的時候檢測是否超時。

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

超時檢測線程

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

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 結(jié)果 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(); 
                // 結(jié)果設(shè)置為超時，從請求 map 中移除 
                responseMap.putIfAbsent(key, RpcResponseFactory.timeout()); 
                requestMap.remove(key); 
            } 
        } 
    } 
 
 
}

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

線程啟動

在 DefaultInvokeService 初始化時啟動：

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

DefaultInvokeService

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

設(shè)置請求時間

•添加請求 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; 
}

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

•添加響應(yīng) addResponse

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

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

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

@Override 
public InvokeService addResponse(String seqId, RpcResponse rpcResponse) { 
    // 1. 判斷是否有效 
    Long expireTime = this.requestMap.get(seqId); 
    // 如果為空，可能是這個結(jié)果已經(jīng)超時了，被定時 job 移除之后，響應(yīng)結(jié)果才過來。直接忽略 
    if(ObjectUtil.isNull(expireTime)) { 
        return this; 
    } 
 
 
    //2. 判斷是否超時 
    if(Times.time() > expireTime) { 
        LOG.info("[Client] seqId:{} 信息已超時，直接返回超時結(jié)果。", seqId); 
        rpcResponse = RpcResponseFactory.timeout(); 
    } 
 
 
    // 這里放入之前，可以添加判斷。 
    // 如果 seqId 必須處理請求集合中，才允許放入。或者直接忽略丟棄。 
    // 通知所有等待方 
    responseMap.putIfAbsent(seqId, rpcResponse); 
    LOG.info("[Client] 獲取結(jié)果信息，seqId: {}, rpcResponse: {}", seqId, rpcResponse); 
    LOG.info("[Client] seqId:{} 信息已經(jīng)放入，通知所有等待方", seqId); 
    // 移除對應(yīng)的 requestMap 
    requestMap.remove(seqId); 
    LOG.info("[Client] seqId:{} remove from request map", seqId); 
    synchronized (this) { 
        this.notifyAll(); 
    } 
    return this; 
}

獲取請求結(jié)果

•獲取相應(yīng) getResponse

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

2.否則進入等待。

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

@Override 
public RpcResponse getResponse(String seqId) { 
    try { 
        RpcResponse rpcResponse = this.responseMap.get(seqId); 
        if(ObjectUtil.isNotNull(rpcResponse)) { 
            LOG.info("[Client] seq {} 對應(yīng)結(jié)果已經(jīng)獲取: {}", seqId, rpcResponse); 
            return rpcResponse; 
        } 
        // 進入等待 
        while (rpcResponse == null) { 
            LOG.info("[Client] seq {} 對應(yīng)結(jié)果為空，進入等待", seqId); 
            // 同步等待鎖 
            synchronized (this) { 
                this.wait(); 
            } 
            rpcResponse = this.responseMap.get(seqId); 
            LOG.info("[Client] seq {} 對應(yīng)結(jié)果已經(jīng)獲取: {}", seqId, rpcResponse); 
        } 
        return rpcResponse; 
    } catch (InterruptedException e) { 
        throw new RpcRuntimeException(e); 
    } 
}

可以發(fā)現(xiàn)獲取部分的邏輯沒變，因為超時會返回一個超時對象：RpcResponseFactory.timeout();

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

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; 
 
 
/** 
 * 響應(yīng)工廠類 
 * @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()); 
    } 
 
 
    /** 
     * 獲取超時響應(yīng)結(jié)果 
     * @return 響應(yīng)結(jié)果 
     * @since 0.0.7 
     */ 
    public static RpcResponse timeout() { 
        return TIMEOUT; 
    } 
 
 
}

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

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

測試代碼

服務(wù)端

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

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); 
    } 
 
 
}

客戶端

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

public static void main(String[] args) { 
    // 服務(wù)配置信息 
    ReferenceConfig<CalculatorService> config = new DefaultReferenceConfig<CalculatorService>(); 
    config.serviceId(ServiceIdConst.CALC); 
    config.serviceInterface(CalculatorService.class); 
    config.addresses("localhost:9527"); 
    // 設(shè)置超時時間為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 服務(wù)開始啟動客戶端 
... 
[INFO] [2021-10-05 14:59:42.504] [main] [c.g.h.r.c.c.RpcClient.connect] - RPC 服務(wù)啟動客戶端完成，監(jiān)聽地址 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 信息已超時，直接返回超時結(jié)果。 
[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} 
[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)放入，通知所有等待方 
[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 對應(yīng)結(jié)果已經(jīng)獲取: DefaultRpcResponse{seqId='null', error=com.github.houbb.rpc.common.exception.RpcTimeoutException, result=null} 
...

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

不足之處

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

責(zé)任編輯：姜華來源：今日頭條