源码剖析 | 蚂蚁金服 mPaaS 框架下的 RPC 调用历程

mPaaS-RPC 是支付宝原生的 RPC 调用库。

在客户端开发过程中，只需要简单的调用库里封装的函数即可完成一次数据请求过程，但是遇到异常情况往往会面对各种异常码却不知所云。所以这篇文章带领大家了解一下 mPaaS-RPC 的调用过程以及各种异常码产生的原因。

1. 使用方法

在 Android 端，RPC 的调用是很简单的，大概分为以下几个过程：

1.1 定义 RPC 接口

首先需要定义 RPC 接口，包括接口名，接口特性（是否需要登录，是否需要签名等），接口的定义方式如下：

public interface LegoRpcService {
    @CheckLogin
    @SignCheck
    @OperationType("alipay.lego.card.style")
    LegoCardPB viewCard(MidPageCardRequestPB var1);
}
复制代码

当客户端调用 viewCard 方法时，框架就会请求 alipay.lego.card.style 对应的 RPC 接口，接口的参数定义在 MidPageCardRequestPB 中。

这个接口调用前需要检查是否登录以及客户端签名是否正确，因此如此简单的一个接口就定义了一个 RPC 请求。

1.2 调用 RPC 请求

定义完接口后，需要调用 RPC 请求，调用方法如下：

//创建参数对象
MidPageCardRequestPB request = new MidPageCardRequestPB();
//填充参数
...

//获取 RpcService
RpcService rpcService = (RpcService)LauncherApplicationAgent.getInstance().getMicroApplicationContext().findServiceByInterface(RpcService.class.getName());
//获取Rpc请求代理类
LegoRpcService service = rpcService.getRpcProxy(LegoRpcService.class);
//调用方法
LegoCardPB result = service.viewFooter(request);
复制代码

调用过程大概就分为以上几步。

值得注意的是，整个过程中我们并没有去实现 LegoRpcService 这个接口，而是通过 rpcService.getRpcProxy 获取了一个代理，这里用的了 Java 动态代理的概念，后面的内容会涉及。

2. 源码解析

接下来我们就来看看这套框架是怎么运行的。

2.1 创建参数对象

参数对象是一个 PB 的对象，这个对象的序列化和反序列化过程需要和服务端对应起来。简单来说，就是这个参数在客户端序列化，作为请求的参数发送请求，然后服务端收到请求后反序列化，根据参数执行请求，返回结果。

MidPageCardRequestPB request = new MidPageCardRequestPB();
复制代码

2.2 获取 RPCService

RPCService 的具体实现是 mpaas-commonservice-git Bundle 中的 RPCServiceImpl

这个添加的过程是在 mPaaS 启动时，调用 CommonServiceLoadAgent 的 load 方法。

@Override
    public final void load() {
        ...
        registerLazyService(RpcService.class.getName(), RpcServiceImpl.class.getName());
        ...
    }
复制代码

RpcServiceImpl 中 getRpcProxy 方法调用的是 RpcFactory 的 getRpcProxy 方法。

@Override
    public <T> T getRpcProxy(Class<T> clazz) {
        return mRpcFactory.getRpcProxy(clazz);
    }
复制代码

2.3 获取 RPC 请求代理类

mRpcFactory 这个对象在 mPaas-Rpc Bundle 中。

public <T> T getRpcProxy(Class<T> clazz) {
        LogCatUtil.info("RpcFactory","clazz=["+clazz.getName()+"]");
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[] { clazz},
            new RpcInvocationHandler(mConfig,clazz, mRpcInvoker));
    }
复制代码

这里就是根据接口创建动态代理的过程，这是 Java 原生支持的一个特性。

简单来说，动态代理即 JVM 会根据接口生成一个代理类，调用接口方法时，会先调用代理类的 invoke 方法，在 invoke 方法中你可以根据实际情况来做操作从而实现动态代理的作用。

2.4 调用方法

动态代理类便调用 RpcInvocationHandler 方法： 即调用定义的 RPC 接口时，会调用到 RpcInvocationHandler 的 invoke 方法：

@Override
    public Object invoke(Object proxy, Method method, Object[] args) throws RpcException {
        return mRpcInvoker.invoke(proxy, mClazz, method, args, buildRpcInvokeContext(method));
    }
复制代码

invoke 方法调用的是 mRpcInvoker 的 invoke 方法，而 mRpcInvoker 是创建 RpcInvocationHandler 传过来的。invoke 方法在原来参数的基础上又传递了 mClazz 和 buildRpcInvokeContext(method) ：

mClazz 很好理解，因为 RpcInvocationHandler 是对应某个类，而 mRpcInvoker 是个单例：它并不知道是在代理哪个类的方法，所以需要明确告诉它。

buildRpcInvokeContext(method) 从名字来看是一个上下文对象，这里面保存了请求的上下文信息。

接下来就到了 RpcInvoker 的 invoke 方法：介绍一下 RPC 框架的设计理念，在 invoke 方法中只定义流程，不做具体操作。框架中会注册很多 Interceptor，每个流程交给都交给 Interceptor 去做。这个思想在网络层的设计架构上很常见，比如很著名的 Spring。遗憾的是，我觉得这套 RPC 框架在后期开发过程中，这个思想有点稍稍被打乱，而 invoke 方法针对一些细节进行了优化处理。

言归正传，回到这个 invoke 方法，我总结了一些调用流程，如下图所示：

简单来说，即依次调用了 preHandle，singleCall 和 postHandle。下面我们来看下代码：

public Object invoke(Object proxy, Class<?> clazz, Method method,
                         Object[] args, InnerRpcInvokeContext invokeContext) throws RpcException {
            ...
            preHandle(proxy, clazz, method, args, method.getAnnotations(),invokeContext);// 前置拦截
           ...
          try{
                response = singleCall(method, args, RpcInvokerUtil.getOperationTypeValue(method, args), id, invokeContext,protoDesc);
                returnObj = processResponse(method,response,protoDesc);
        } catch (RpcException exception) {
            exceptionHandle(proxy, response!=null?response.getResData():null, clazz, method, args, method.getAnnotations(), exception, invokeContext);// 异常拦截
        }
          ...
        postHandle(proxy, response!=null?response.getResData():null, clazz, method, args, method.getAnnotations(), invokeContext);// 后置拦截
         ...
        return returnObj;
    }
复制代码

2.5 前置拦截

首先来看下前置拦截 preHandle 方法：

private void preHandle(final Object proxy, final Class<?> clazz, final Method method,
                           final Object[] args, Annotation[] annotations, InnerRpcInvokeContext invokeContext) throws RpcException {
        handleAnnotations(annotations, new Handle() {
            @Override
            public boolean handle(RpcInterceptor rpcInterceptor, Annotation annotation)
                    throws RpcException {
                if (!rpcInterceptor.preHandle(proxy, RETURN_VALUE, new byte[]{}, clazz, method,
                        args, annotation, EXT_PARAM)) {
                    throw new RpcException(RpcException.ErrorCode.CLIENT_HANDLE_ERROR,
                            rpcInterceptor + "preHandle stop this call.");
                }
                return true;
            }
        });

        RpcInvokerUtil.preHandleForBizInterceptor(proxy, clazz, method, args, invokeContext, EXT_PARAM, RETURN_VALUE);

        //mock RPC 限流
        RpcInvokerUtil.mockRpcLimit(mRpcFactory.getContext(),method, args);
    }
复制代码

前置拦截有三个过程：首先 handleAnnotations 处理方法的注解，然后执行业务层定义的拦截器，最后模拟 RPC 限流。

2.6 处理注解

这是前置拦截中最重要的一步，主要调用了 handleAnnotations 方法。handleAnnotations 方法给了一个回调，回调的参数是注解以及对应的 RpcInterceptor，返回后调用 RpcInterceptor 的 preHandle 方法。

介绍 handleAnnotations 之前先简单说一下 RpcInterceptor，这个在框架中叫做拦截器：

public interface RpcInterceptor {
    public boolean preHandle(Object proxy,ThreadLocal<Object> retValue,  byte[] retRawValue, Class<?> clazz, Method method, Object[] args, Annotation annotation,ThreadLocal<Map<String,Object>> extParams) throws RpcException;

    public boolean postHandle(Object proxy,ThreadLocal<Object> retValue,  byte[] retRawValue, Class<?> clazz, Method method, Object[] args, Annotation annotation)
                                                                                            throws RpcException;

    public boolean exceptionHandle(Object proxy,ThreadLocal<Object> retValue,  byte[] retRawValue, Class<?> clazz, Method method, Object[] args,
                                   RpcException exception, Annotation annotation) throws RpcException;
}
复制代码

简单来说，程序一开始会注册几个拦截器，每个拦截器对应一种注解。invoke 方法中处理注解的时候，会找到对应的拦截器，然后调用拦截器相应的方法。正如前面流程图中所说的，如果拦截器返回 true，继续往下运行，如果返回 false，则抛出相应的异常。

接下来继续说 handleAnnotations： handleAnnotations 其实就是查找拦截器的方法，看一下具体实现：

private boolean handleAnnotations(Annotation[] annotations, Handle handle) throws RpcException {
            for (Annotation annotation : annotations) {
                Class<? extends Annotation> c = annotation.annotationType();
                RpcInterceptor rpcInterceptor = mRpcFactory.findRpcInterceptor(c);
                ret = handle.handle(rpcInterceptor, annotation);

            }
}
复制代码

我们调用了 mRpcFactory.findRpcInterceptor(c) 方法去查找拦截器： mRpcFactory.findRpcInterceptor(c) 查找了两个地方：

• 一个是 mInterceptors；
• 另一个是 GLOBLE_INTERCEPTORS 里。

public RpcInterceptor findRpcInterceptor(Class<? extends Annotation> clazz) {
        RpcInterceptor rpcInterceptor = mInterceptors.get(clazz);
        if (rpcInterceptor != null) {
            return rpcInterceptor;
        }
        return GLOBLE_INTERCEPTORS.get(clazz);
    }
复制代码

这两个地方的拦截器其实是一样的，因为 addRpcInterceptor 的时候会往这两个地方都添加一次。

public void addRpcInterceptor(Class<? extends Annotation> clazz, RpcInterceptor rpcInterceptor) {
        mInterceptors.put(clazz, rpcInterceptor);
        addGlobelRpcInterceptor(clazz,rpcInterceptor);
    }
复制代码

而之前提到的 Spring 的处理方式是：每种情况，多个拦截器依次处理，扩展性比较好。

这里拦截器的添加是在 commonbiz Bundle 中 CommonServiceLoadAgent 的 afterBootLoad 方法中，这个方法也是在 mPaaS 框架启动的时候调用的。

rpcService.addRpcInterceptor(CheckLogin.class, new LoginInterceptor());
rpcService.addRpcInterceptor(OperationType.class, new CommonInterceptor());
rpcService.addRpcInterceptor(UpdateDeviceInfo.class, new CtuInterceptor(mMicroAppContext.getApplicationContext()));
复制代码

一共针对三种注解添加了三个拦截器，在文章的最后我们分析一下每种拦截器做了什么操作：

handleAnnotations 中找到了对应的拦截器，就调用其 preHandle 方法进行前置拦截；

preHandle 中处理完注解的前置拦截，会在 preHandleForBizInterceptor 处理上下文对象中带的拦截器；

上下文对象中的拦截器和注解拦截器道理是一样的，这个阶段我看了一下上下文对象 Context 中并没有设置拦截器。如果有的话，即取出来然后依次调用对应方法。

preHandle 的最后一步是模拟网络限流，这是在测试中使用的，如果测试打开 RPC 限流功能，那么这里会限制 RPC 的访问来模拟限流的情况。

2.7 singlecall

处理完前置拦截，又回到 RpcInvoker 的 invoke 方法，接下来会调用 singleCall 去发起网络请求。

private Response singleCall(Method method, Object[] args,
                                String operationTypeValue, int id, InnerRpcInvokeContext invokeContext,RPCProtoDesc protoDesc) throws RpcException {
        checkLogin(method,invokeContext);
        Serializer serializer = getSerializer(method, args, operationTypeValue,id,invokeContext,protoDesc);
    if (EXT_PARAM.get() != null) {
        serializer.setExtParam(EXT_PARAM.get());
    }
    byte[] body = serializer.packet();
    HttpCaller caller = new HttpCaller(mRpcFactory.getConfig(), method, id, operationTypeValue, body,
    serializerFactory.getContentType(protoDesc), mRpcFactory.getContext(),invokeContext);
        addInfo2Caller(method, serializer, caller, operationTypeValue, body, invokeContext);
        Response response = (Response) caller.call();// 同步
        return response;
    }
复制代码

singleCall 一开始又 checkLogin，接着根据参数的类型获取了序列化器 Serializer，并进行了参数的序列化操作，序列化后的参数作为整个请求的 body。然后一个 HttpCaller 被创建，HttpCaller 以下是网络传输层的封装代码，这篇文章中暂时不关注。

在实际发送请求之前，需要调用 addInfo2Caller 往 HttpCaller 中添加一些通用信息，比如序列化版本，contentType，时间戳，还有根据 SignCheck 注解决定要不要在请求里添加签名信息。

按照我的理解，其实这块也应该在各个 Intercepter 中处理，不然把 SignCheck 这个注解单独拿出来处理，代码实在是不好看。

最后我们可以去实际调用 caller.call 方法发送请求了，然后收到服务端的回复。

2.8 ProcessResponse

这样就执行完 singleCall 方法，获得了服务端的回复。

过程又回到了 invoke 里，获得服务端 response 后，调用 processResponse 去处理这个回复。处理回复的过程其实就是把服务端的返回结果反序列化，是前面发送请求的一个逆过程，代码如下：

private Object processResponse(Method method,
                                 Response response,RPCProtoDesc protoDesc) {
        Type retType = method.getGenericReturnType();
        Deserializer deserializer = this.serializerFactory.getDeserializer(retType,response,protoDesc);
        Object object = deserializer.parser();
        if (retType != Void.TYPE) {// 非void
            RETURN_VALUE.set(object);
        }
        return object;
    }
复制代码

在 preHandle，singleCall 和 processResponse 这三个过程中，如果有 RpcException 抛出（处理过程中所有的异常情况都是以 RpcException 的形式抛出），invoke 中会调用 exceptionHandle 异常。

2.9 异常处理

exceptionHandle 也是同样地去三个 Interceptor 中找相应注释的拦截器，并调用 exceptionHandle：如果返回 true，说明需要继续处理，再把这个异常抛出去，交给业务方处理；如果返回 false，则代表异常被吃掉了，不需要被继续处理了。

private void exceptionHandle(final Object proxy, final byte[] rawResult, final Class<?> clazz,
                                 final Method method, final Object[] args,
                                 Annotation[] annotations, final RpcException exception,InnerRpcInvokeContext invokeContext)
            throws RpcException {
        boolean processed = handleAnnotations(annotations, new Handle() {
            @Override
            public boolean handle(RpcInterceptor rpcInterceptor, Annotation annotation)
                    throws RpcException {
                if (rpcInterceptor.exceptionHandle(proxy, RETURN_VALUE, rawResult, clazz, method,
                        args, exception, annotation)) {
                    LogCatUtil.error(TAG, exception + " need process");
                    // throw exception;
                    return true;
                } else {
                    LogCatUtil.error(TAG, exception + " need not process");
                    return false;
                }
            }
        });
        if (processed) {
            throw exception;
        }

    }
复制代码

2.10 后置处理

处理完异常之后，invoke 方法继续往下执行，下一步是调用 postHandle 进行后置拦截。流程跟前置拦截完全一样，先是去找三个默认的拦截器处理，然后再去 invokeContext 去找业务定制的拦截器，目前这一块没有任何实现。

处理 preHandle，singeCall，exceptionHandle 和 postHandle 这几个主要的流程，invoke 会调用 asyncNotifyRpcHeaderUpdateEvent 去通知关心 Response Header 的 Listener，然后打印返回结果的信息，之后整个流程结束，返回请求结果。

3. 默认拦截器实现

默认拦截器一共有三个，针对三种不同的注解：

rpcService.addRpcInterceptor(CheckLogin.class, new LoginInterceptor());
rpcService.addRpcInterceptor(OperationType.class, new CommonInterceptor());
rpcService.addRpcInterceptor(UpdateDeviceInfo.class, new CtuInterceptor(mMicroAppContext.getApplicationContext()));
复制代码

3.1 LoginInterceptor

第一个 LoginInterceptor，顾名思义，就是检查登录的拦截器。这里我们只实现了前置拦截的方法 preHandle：

public boolean preHandle(Object proxy, ThreadLocal<Object> retValue, byte[] retRawValue, Class<?> clazz,
                             Method method, Object[] args, Annotation annotation,ThreadLocal<Map<String,Object>> extParams) throws RpcException {
        AuthService authService = AlipayApplication.getInstance().getMicroApplicationContext().getExtServiceByInterface(AuthService.class.getName());
        if (!authService.isLogin() && !ActivityHelper.isBackgroundRunning()) {//未登录
            LoggerFactory.getTraceLogger().debug("LoginInterceptor", "start login:" + System.currentTimeMillis());

            Bundle params = prepareParams(annotation);
            checkLogin(params);
            LoggerFactory.getTraceLogger().debug("LoginInterceptor", "finish login:" + System.currentTimeMillis());
            fail(authService);
        }
        return true;
    }
复制代码

检查是否登录：

• 如果没登录，抛出 CLIENT_LOGIN_FAIL_ERROR = 11 则异常。

3.2 CommonInterceptor

通用拦截器，拦截的是 OperationType 注解，这个注解的 value 是 RPC 请求的方法名，所以可以看出 CommonInterceptor 会处理所有的 RPC 请求，这也是为什么叫 CommonInterceptor。

public boolean preHandle(Object proxy, ThreadLocal<Object> retValue, byte[] retRawValue, Class<?> clazz,
                             Method method, Object[] args, Annotation annotation, ThreadLocal<Map<String, Object>> extParams)
            throws RpcException {
        checkWhiteList(method, args);
        checkThrottle();
        ...
        writeMonitorLog(ACTION_STATUS_RPC_REQUEST, clazz, method, args);

        for (RpcInterceptor i : RpcCommonInterceptorManager.getInstance().getInterceptors()) {
            i.preHandle(proxy, retValue, retRawValue, clazz, method, args, annotation, extParams);
        }

        return true;
    }
复制代码

• 第一步：检查白名单：

在启动的前3s内，为了保证性能，只有白名单的请求才能发送。如果不在白名单，会抛出 CLIENT_NOTIN_WHITELIST = 17

• 第二步：检查是否限流：

限流是服务端设定的，如果服务端设置了限流，则在某次请求的时候，服务端会返回 SERVER_INVOKEEXCEEDLIMIT=1002异常，这时候 CommonInterceptor 会从返回结果中取到限流到期时间

if(exception.getCode() == RpcException.ErrorCode.SERVER_INVOKEEXCEEDLIMIT){
            String control = exception.getControl();
            if(control!=null){
                mWrite.lock();
                try{
                    JSONObject jsonObject = new JSONObject(control);
                    if(jsonObject.getString("tag").equalsIgnoreCase("overflow")){
                        mThrottleMsg = exception.getMsg();
                        mControl = control;

                        //如果是own的异常，需要更新限流结束的时间
                        if ( exception.isControlOwn() ){
                            mEndTime = System.currentTimeMillis()+jsonObject.getInt("waittime")*1000;
                        }
                    }
                }
        }
复制代码

• 第三步：写监控日志

writeMonitorLog(ACTION_STATUS_RPC_REQUEST, clazz, method, args);
复制代码

• 第四步：处理业务定制的拦截器

for (RpcInterceptor i : RpcCommonInterceptorManager.getInstance().getInterceptors()) {
            i.preHandle(proxy, retValue, retRawValue, clazz, method, args, annotation, extParams);
        }
复制代码

前面我们说过 RPC 框架中一个拦截器是跟一个注解绑定的，比如 CommonIntercetor 是跟 operatorType 注解绑定的。但是如果业务方想定制 operatorType 注解的拦截器怎么办，就需要在 CommonIntercetor 下面再绑定拦截器列表。目前这里没有实现，可以忽略。

异常拦截 exceptionHandle 是用来处理服务端返回结果中的异常情况的，业务方可以根据自己的服务端返回结果进行定制。举个例子，假如你本地的 session 失效了，请求服务端结果后，服务端返回了登录失效的状态码 SESSIONSTATUS_FAIL 。收到这个异常后业务方可以进行相应的处理了，比如是否需要使用本地保存的账号密码进行自动登录，或者弹出登录框请求用户登录，或者直接返回让业务方处理等等各种形式。

后置拦截 postHandle 做了一件事，记录了服务端返回结果的日志。

3.3 CtuInterceptor

CtuInterceptor 拦截器对应的是 UpdateDeviceInfo 这个注解。这个注解表示这个 RPC 请求需要设备信息。所以前置拦截的时候将设备信息写入请求的参数里面。

RpcDeviceInfo rpcDeviceInfo = new RpcDeviceInfo();
        DeviceInfo deviceInfo = DeviceInfo.getInstance();
        // 添加一些设备信息到deviceInfo
        ....
复制代码

exceptionHandle 和 postHandle 都没有处理。

以上就是系统默认的三个拦截器，和整个 mPaaS-RPC Bundle 中进行的流程。其实这样看起来 mPaaS-RPC 只负责网络请求的封装和发送，整个流程还是很简单的。然而网络请求返回后根据不同的错误码进行不同的处理才是真正复杂的部分，这部分本来是交给具体业务方去处理的。

不过良心支付宝又提供了一层封装 RPC-Beehive 组件，这层是在网络层框架和业务方之间的一层封装，将通用的一些异常码进行了处理，比如，请求是转菊花，或者返回异常后显示通用异常界面。

以上便是针对 mPaaS-RPC 的源码剖析，欢迎大家反馈想法或建议，一起讨论修正。

关注我们公众号，获得第一手 mPaaS 技术实践干货