手写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<String, GenericService> 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<Service> 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<User> getUser(Long id);
String getUserString(Long id);
}
- 创建SpringBoot工程consumer-v2，并添加ship-rpc-spring-boot-starter依赖
<dependency>
<groupId>io.github.2ysp</groupId>
<artifactId>ship-rpc-spring-boot-starter</artifactId>
<version>1.0.1-RELEASE</version>
</dependency>
- 编写泛化调用测试接口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
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 read
Requests/sec: 3689.17
Transfer/sec: 728.30KB
可以看到QPS大概能达到3600多，还是不错的。
五、总结
希望本篇文章能帮助你了解泛化调用，这次除了增加了泛化调用的功能外，还对以前的代码进行了重构，包括增加Nacos注册中心支持，增加hessian序列化协议，包结构优化等，后面有时间会该框架增加更多功能。
参考：