手写RPC框架之泛化调用

一、背景

前段时间了解了泛化调用这个玩意儿，又想到自己之前写过一个RPC框架（参考《手写一个RPC框架》），于是便想小试牛刀。

二、泛化调用简介

什么是泛化调用

(资料图)

泛化调用就是在不依赖服务方接口jar包的情况下进行调用，包括对调用方法的泛化、参数的泛化和返回值的泛化。

泛化调用的使用场景

常规的PRC调用都是客户端依赖服务端提供的接口jar包，然后利用动态代理技术，像调用本地方法一样调用远程方法，但是有些场景下客户端无法依赖jar包，也要调用远程方法，这时就需要用到泛化调用了。

常规的使用场景包括：

网关，如果服务内部调用使用RPC协议，对外暴露HTTP接口，这时就需要在网关做协议转换（HTTP转RPC协议），但是网关不可能依赖所有接口的jar包，只能采用泛化调用。

测试平台

实现方案

实现方案有两种:

第一种是基于Java Bean的泛化调用，例如dubbo的泛化调用会将参数转换成JavaBeanDescriptor，代码可以参考GenericFilter。第二种是基于序列化中间体的泛化调用，如sofa-rpc，使用了sofa-hessian序列化框架，sofa-hessian是在hessian序列化框架基础上进行二次开发的，抽象出了序列化中间体，如GenericObject、GenericMap、GenericArray等。三、开发实现3.1 类图

客户端

服务注册和发现

3.2 数据传输过程3.3 客户端实现

首先定义一个泛化调用接口GenericService

public interface GenericService {    /**     * 泛化调用     * @param methodName     * @param parameterTypeNames     * @param args     * @return     */    Object $invoke(String methodName, String[] parameterTypeNames, Object[] args);}

注意：这里参数类型parameterTypeNames用的是参数类型名称数组而不是Class数组，是因为泛化调用客户端可能不存在对应的类。

默认实现类DefaultGenericService

/** * @Author: Ship * @Description: * @Date: Created in 2023/6/15 */public class DefaultGenericService implements GenericService {    private MethodInvoker methodInvoker;    private String interfaceClassName;    public DefaultGenericService(MethodInvoker methodInvoker, String interfaceClassName) {        this.methodInvoker = methodInvoker;        this.interfaceClassName = interfaceClassName;    }    @Override    public Object $invoke(String methodName, String[] parameterTypeNames, Object[] args) {        return methodInvoker.$invoke(interfaceClassName, methodName, parameterTypeNames, args, true);    }}

因为DefaultGenericService是接口维度的，所以我们还需要一个工厂类去创建它的实例，同时为了避免重复创建对象，还要缓存接口维度的实例（享元模式）。

/** * @Author: Ship * @Description: * @Date: Created in 2023/6/15 */public final class GenericServiceFactory {    /**     * 实例缓存，key:接口类名     */    private static final Map INSTANCE_MAP = new ConcurrentHashMap<>();    private GenericServiceFactory() {}    /**     * @param interfaceClassName     * @return     */    public static GenericService getInstance(String interfaceClassName) {        return INSTANCE_MAP.computeIfAbsent(interfaceClassName, clz -> {            MethodInvoker methodInvoker = SpringContextHolder.getBean(MethodInvoker.class);            DefaultGenericService genericService = new DefaultGenericService(methodInvoker, interfaceClassName);            return genericService;        });    }}

MethodInvoker维护了客户端调用服务端的核心逻辑，同时兼容泛化调用和普通RPC调用这两种调用方式。

实现类DefaultMethodInvoker

public class DefaultMethodInvoker implements MethodInvoker {    private ServerDiscoveryManager serverDiscoveryManager;    private NetClient netClient;    private LoadBalance loadBalance;    public DefaultMethodInvoker(ServerDiscoveryManager serverDiscoveryManager, NetClient netClient, LoadBalance loadBalance) {        this.serverDiscoveryManager = serverDiscoveryManager;        this.netClient = netClient;        this.loadBalance = loadBalance;    }    @Override    public Object $invoke(String interfaceClassName, String methodName, String[] parameterTypeNames, Object[] args, Boolean generic) {        // 1.获得服务信息        String serviceName = interfaceClassName;        List services = serverDiscoveryManager.getServiceList(serviceName);        Service service = loadBalance.chooseOne(services);        // 2.构造request对象        RpcRequest request = new RpcRequest();        request.setRequestId(UUID.randomUUID().toString());        request.setServiceName(service.getName());        request.setMethod(methodName);        request.setParameters(args);        request.setParameterTypeNames(parameterTypeNames);        request.setGeneric(generic);        // 3.协议层编组        MessageProtocol messageProtocol = MessageProtocolsManager.get(service.getProtocol());        RpcResponse response = netClient.sendRequest(request, service, messageProtocol);        if (response == null) {            throw new RpcException("the response is null");        }        // 6.结果处理        if (RpcStatusEnum.ERROR.getCode().equals(response.getRpcStatus())) {            throw response.getException();        }        if (RpcStatusEnum.NOT_FOUND.getCode().equals(response.getRpcStatus())) {            throw new RpcException(" service not found!");        }        return response.getReturnValue();    }}

3.4 服务端实现

RequestHandler的核心逻辑就是利用反射调用对应的方法

public class RequestHandler {    private MessageProtocol protocol;    private ServerRegister serverRegister;    public RequestHandler(MessageProtocol protocol, ServerRegister serverRegister) {        this.protocol = protocol;        this.serverRegister = serverRegister;    }    public byte[] handleRequest(byte[] data) throws Exception {        // 1.解组消息        RpcRequest req = this.protocol.unmarshallingRequest(data);        // 2.查找服务对应        ServiceObject so = serverRegister.getServiceObject(req.getServiceName());        RpcResponse response = null;        if (so == null) {            response = new RpcResponse(RpcStatusEnum.NOT_FOUND);        } else {            try {                // 3.反射调用对应的方法过程                Method method = so.getClazz().getMethod(req.getMethod(), ReflectUtils.convertToParameterTypes(req.getParameterTypeNames()));                Object returnValue = method.invoke(so.getObj(), req.getParameters());                response = new RpcResponse(RpcStatusEnum.SUCCESS);                if (req.getGeneric()) {                    response.setReturnValue(RpcResponseUtils.handlerReturnValue(returnValue));                } else {                    response.setReturnValue(returnValue);                }            } catch (Exception e) {                response = new RpcResponse(RpcStatusEnum.ERROR);                String errMsg = JSON.toJSONString(e);                response.setException(new RpcException(errMsg));            }        }        // 编组响应消息        response.setRequestId(req.getRequestId());        return this.protocol.marshallingResponse(response);    }}

可以看到这里针对泛化调用的返回值作了特殊处理，因为如果返回的是POJO对象的话客户端是没有对应的类的，那么如何泛化处理呢？

分了三种情况处理：

如果是JDK的基本类型包装类，如Long、Integer则直接不处理返回。如果是原始类型如int、long，则报错不支持。如果是POJO自定义对象，则转换成Map返回给客户端。

服务注册和发现部分的代码就不贴了，有兴趣可以自行查看，代码地址。

四、测试4.1 功能测试服务端provider项目提供两个根据id查询用户的接口，如下

public interface UserService {    ApiResult getUser(Long id);    String getUserString(Long id);}

创建SpringBoot工程consumer-v2，并添加ship-rpc-spring-boot-starter依赖

            io.github.2ysp            ship-rpc-spring-boot-starter            1.0.1-RELEASE        

编写泛化调用测试接口GenericTestController

@RestController@RequestMapping("/GenericTest")public class GenericTestController {    @GetMapping("/user")    public String getUserString(@RequestParam("id") Long id) {        //cn.sp.UserService.getUserString        GenericService instance = GenericServiceFactory.getInstance("cn.sp.UserService");        Object result = instance.$invoke("getUserString", new String[]{"java.lang.Long"}, new Object[]{id});        return result.toString();    }    @GetMapping("")    public String getUser(@RequestParam("id") Long id) {        //cn.sp.UserService.getUser        GenericService instance = GenericServiceFactory.getInstance("cn.sp.UserService");        Object result = instance.$invoke("getUser", new String[]{"java.lang.Long"}, new Object[]{id});        return result.toString();    }}

本地依次启动nacos，provider和consumer-v2工程

控制台能看到注册的服务说明provider启动成功。

postman请求接口http://localhost:8081/GenericTest/user?id=1，返回如下说明调通了

{"code":200,"data":{"gender":2,"id":1,"name":"XX","webSite":"www.aa.com"},"message":"success"}

然后在cn.sp.GenericTestController#getUser方法打断点，再请求接口http://localhost:8081/GenericTest?id=1

可以看出接口正确返回了，并且把ApiResult对象转换成了Map。

4.2 压测

压测环境：

MacBook Pro 13英寸

处理器 2.3 GHz 四核Intel Core i7

内存 16 GB 3733 MHz LPDDR4X

一个生产者一个消费者

压测工具：

wrk

压测命令：

wrk -c 100 -t 20 -d 10s http://localhost:8081/GenericTest?id=1

用100个链接，20个线程压测10秒钟

压测结果：

Running 10s test @ http://localhost:8081/GenericTest?id=1  20 threads and 100 connections  Thread Stats   Avg      Stdev     Max   +/- Stdev    Latency    28.12ms   19.11ms 175.97ms   71.66%    Req/Sec   185.58     35.41   272.00     78.75%  37000 requests in 10.03s, 7.13MB readRequests/sec:   3689.17Transfer/sec:    728.30KB

可以看到QPS大概能达到3600多，还是不错的。

五、总结

希望本篇文章能帮助你了解泛化调用，这次除了增加了泛化调用的功能外，还对以前的代码进行了重构，包括增加Nacos注册中心支持，增加hessian序列化协议，包结构优化等，后面有时间会该框架增加更多功能。

参考：

RPC框架泛化调用原理及转转的实践https://nacos.io/zh-cn/docs/sdk.html

关键词：