rpc回调的几种设计方式

Posted 2021-06-28 littleschemer

传统的http是基于请求-响应模式。也就是说，客户端（浏览器）主动发起请求，服务器才会给予返回值。

类似于http，RPC（RemoteProcedureCall）的远程方法调用的返回值，也是一种请求-响应模式。RPC提供类似于下面的API来获取方法的返回值。

public Response request(Message req) {
      // do sth
    return new Response();
}

本文主要是探讨rpc回调函数的三种设计模式

1. 基于同步阻塞的模式

第一种方式，也是最直观的模式，发起方调用API的时候，会同步阻塞当前线程直到获取响应消息。当然，为了处理各种突发事件，会设置超时参数。下面先给一个demo，然后再讨论逻辑过程。

 /**
     * 发送消息并返回执行结果
     * @param session
     * @param request
     * @return
     */
    public Message request(Session session, G2FCallBack request) throws InterruptedException, CallTimeoutException {
        int timeout = 5000;
        int index = request.getIndex();
        session.sendMessage(request);
        final RequestResponseFuture future = new RequestResponseFuture(index,  timeout,null);
        try {
            CallBackService.getInstance().register(index, future);
            Message responseMessage = future.waitResponseMessage(timeout);
            if (responseMessage == null) {
                CallTimeoutException exception = new CallTimeoutException("send request message  failed");
                future.setCause(exception);
                throw exception;
            } else {
                return responseMessage;
            }
        } catch (InterruptedException e) {
            future.setCause(e);
            throw e;
        } finally {
            CallBackService.getInstance().remove(index);
        }
    }

首先需要说明的一点是，由于请求发起方与函数处理方处于两个不同的进程。为了将两个消息关联在一起。需要请求方带上一个唯一id，这个id不用全局唯一，只需要在发起方内部唯一即可。

算法大概可以分为下面几个步骤：

1. 消息发送之前，生成一个唯一ID，同时注册一个回调接口。将这两者绑定缓存起来

2. 消息发送后，阻塞当前线程，等待响应结果。可以利用CountDownLatch的wait(timeout)方法。

3. 当远程进程返回结果，利用响应包的关联ID找到已注册的回调方法，执行CountDownLatch的countDown()方法，即可唤醒请求方的线程；若在指定时间内无法获得回调消息，则代表请求超时，抛出超时受检异常。

2. 基于异步操作的回调函数

在第一种模式下，以RPC的风格获得回调结果，代码比较优雅。但有一个致命的缺点，执行远程方法的时候，会阻塞当前线程。极端情况下，远程不可达，则每次调用都会阻塞到超时结束。

写过浏览器客户端逻辑的同学，应该对javascript的ajax函数很熟悉。

$.ajax({
    url:"XXXX",
    type: "GET",
    data: {username:$("#username").val()},
    dataType: "json",
    complete:function(msg){ 
        //请求完成后调用的回调函数（请求成功或失败时均调用）
    } , 
    error:function(msg){ 
        //请求失败时被调用的函数 
    } , 
    Sucess:function(msg){ 
        //请求成功后调用的回调函数 
    } 
});

注册回调函数之后，即可在获取服务端返回值的时候自动执行回调函数的逻辑。

类似地，我们的rpc回调也可以设计成ajax的模式，呈现出下面的API

    public void request(Session session, G2FCallBack request, RequestCallback callBack) {
        int timeout = 5000;
        int index = request.getIndex();
        final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(index, timeout, callBack);
        CallBackService.getInstance().register(index, requestResponseFuture);
        session.sendMessage(request);
    }

请求方调用requset()方法后，不会阻塞当前线程。当远程消息返回的时候，自动执行回调函数。

    public void fillCallBack(int index, Message message) {
        RequestResponseFuture future = remove(index);
        if (future == null) {
            LoggerUtils.error("回调信息丢失 msg {}", message);
            return;
        }
        RequestCallback callback = future.getRequestCallback();
        if (callback != null) {
            callback.onSuccess(message);
        }
        if (future != null) {
            future.putResponseMessage(message);
        }
    }

3. 基于Promise的回调函数

第二种模式其实已经很完美了，但还是会有一个瑕疵。熟悉javascript的同学，肯定对“回调地狱”这个名次不陌生，说的是，当javascript使用层层嵌套的回调函数后带来的设计灾难，代码惨不忍睹。于是，Promise应运而生。有了 Promise 对象，就可以将异步操作以同步操作的流程表达出来，避免了层层嵌套的回调函数。

const fs=require('fs')

function getPromise(filename){
    return new Promise((resolve,reject)=>{
    //异步操作读文件
    fs.readFile(`//demo.txt`,'utf-8',(err,data)=>{
        if(err){
            //读取失败
            reject(err)  
        }else{
            //读取成功
            resolve(data)
        }
    })
})
}

getPromise('a').then((data)=>{
    console.log(data)
    return getPromise('b')
}).then((data)=>{
    console.log(data)
    return getPromise('c')
}).then((data)=>{
    console.log(data)
}

Java的线程池虽然提供了异步执行任务的接口，但遗憾的是该方法返回的Future对象，如果想获取执行结果，要么通过轮训检查结果，要么简单粗暴通过调用get()方法阻塞后获得结果。

<T> Future<T> submit(Callable<T> task);

基于此缺点。guava第三方库提供了ListenableFuture接口，Netty提供了PromiseTask接口，本质都是通过监听器避免阻塞调用。其实JDK的CompletableFuture也能实现目的。

public static CompletableFuture<RpcCallbackResponse> sendToCenter(RpcCallbackRequest message) {
        RpcClientRouter clientRouter = GameContext.getBean(RpcClientRouter.class);
        IdSession session = clientRouter.getCenterSession();
      
        int index = idFactory.getAndIncrement();
        message.setCallbackId(index);

        CompletableFuture future = new CompletableFuture();
        Callback callback = new Callback();
        callback.setFuture(future);
        CallbackHandler.registerCallback(index, callback);

        ScheduledFuture timeout = GameContext.getSchedulerManager().schedule(() -> {
            LoggerUtils.error("跨服消息[{}]回调超时", message.getClass().getSimpleName());
            CallbackHandler.removeCallback(index);
            callback.getFuture().completeExceptionally(new RpcTimeoutException());
        }, Callback.TIME_OUT);
        callback.setTimeout(timeout);

        session.sendPacket(message);

        return future;
    }

客户端代码，非常优雅 ^ - ^

RpcReqHello req = new RpcReqHello();
req.setContent("hello, game");

CrossMessageUtil.sendToCenter(req).thenAccept(m -> {
    System.out.println(m);
});

 

总结下三种模式的优缺点

	优点	缺点
同步阻塞	客户端代码调用优雅	会阻塞当前线程
异步回调	异步不阻塞当前线程	嵌套回调代码丑陋
promise	初学者不好理解	嵌套回调代码优雅