easy-RPC 是一款低性能的服务框架,主要用于学习造轮子!
学习本项目可以让你从零开始实现一个类似 Dubbo 服务框架 mini 版RPC。
如果你认真学下来,可以掌握以下的技术:
easy-rpc-spring-boot-starter
,所以会加载spring.factories里面的AutoConfiguration
。AutoConfiguration
会初始化DefaultRpcListener
。DefaultRpcListener
初始化。因为DefaultRpcListener
实现了ApplicationListener<ContextRefreshedEvent>
,所以会重写onApplicationEvent方法,主要初始化了RpcServer和RpcClient。ServiceExpose
的Bean,DefaultRpcListener#registerInstanceInterfaceInfo
,将服务名和对应的实例存放到本地map中,且注册到远程服务器上。NettyRpcServer#start
。该方法主要是用到了Netty,而最核心的是ChannelRequestHandler
,该类实现了ChannelInboundHandlerAdapter
,当有消息进来,会执行channelRead方法。RequestHandler#handleRequest
,根据RpcRequest获取serviceName,再根据serviceName获取ServiceInterfaceInfo,从而获取到对应的Class和实例。DefaultRpcListener#initRpcClient
,找到所有属性带有注解ServiceReference
的Bean,重新设置属性clientProxyFactory.getProxyInstance(fieldClass)
。根据服务名称获取远程实例。NettyRpcClient#sendMessage
。获取到服务端的响应,返回数据。 //获取代理实例
public <T> T getProxyInstance(Class<T> clazz) {
return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 第一步:通过服务发现机制选择一个服务提供者暴露的服务
String serviceName = clazz.getName();
ServiceInterfaceInfo serviceInterfaceInfo = serviceDiscovery.selectOneInstance(serviceName);
logger.info("Rpc server instance list: {}", serviceInterfaceInfo);
if (serviceInterfaceInfo == null) {
throw new RpcException("No rpc servers found.");
}
// 第二步:构造 rpc 请求对象
final RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setServiceName(serviceName);
rpcRequest.setMethodName(method.getName());
rpcRequest.setParameterTypes(method.getParameterTypes());
rpcRequest.setParameters(args);
// 第三步:编码请求消息, TODO: 这里可以配置多种编码方式
byte[] data = messageProtocol.marshallingReqMessage(rpcRequest);
// 第四步:调用 rpc client 开始发送消息
byte[] byteResponse = rpcClient.sendMessage(data, serviceInterfaceInfo);
// 第五步:解码响应消息
final RpcResponse rpcResponse = messageProtocol.unmarshallingRespMessage(byteResponse);
// 第六步:解析返回结果进行处理
if (rpcResponse.getException() != null) {
throw rpcResponse.getException();
}
return rpcResponse.getRetValue();
}
});
}
连接服务端,bootstrap.connect(ip, port).sync();
。服务端的启动方式是ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
。
服务注册与发现。
//ZookeeperServiceRegistry#register
public void register(ServiceInterfaceInfo serviceInterfaceInfo) throws Exception {
logger.info("Registering service: {}", serviceInterfaceInfo);
super.register(serviceInterfaceInfo);
// 创建 ZK 永久节点(服务节点)
String serviceName = serviceInterfaceInfo.getServiceName();
String servicePath = "/com/leixiaoshuai/easyrpc/service/" + serviceName;
if (!zkClient.exists(servicePath)) {
zkClient.createPersistent(servicePath, true);
logger.info("Created node: {}", servicePath);
}
// 创建 ZK 临时节点(实例节点)
String uri = JSON.toJSONString(serviceInterfaceInfo);
uri = URLEncoder.encode(uri, "UTF-8");
String uriPath = servicePath + "/" + uri;
if (zkClient.exists(uriPath)) {
zkClient.delete(uriPath);
}
zkClient.createEphemeral(uriPath);
logger.info("Created ephemeral node: {}", uriPath);
}
// Nacos注册服务
public void register(ServiceInterfaceInfo serviceInterfaceInfo) throws Exception {
super.register(serviceInterfaceInfo);
// 注册当前服务实例
naming.registerInstance(serviceInterfaceInfo.getServiceName(), buildInstance(serviceInterfaceInfo));
}
//Nacos服务发现
public ServiceInterfaceInfo selectOneInstance(String serviceName) {
Instance instance;
try {
// 调用 nacos 提供的接口,随机挑选一个服务实例,负载均衡的算法依赖 nacos 的实现
instance = namingService.selectOneHealthyInstance(serviceName);
} catch (NacosException e) {
log.error("Nacos exception", e);
return null;
}
// 封装实例对象返回
ServiceInterfaceInfo serviceInterfaceInfo = new ServiceInterfaceInfo();
serviceInterfaceInfo.setServiceName(instance.getServiceName());
serviceInterfaceInfo.setIp(instance.getIp());
serviceInterfaceInfo.setPort(instance.getPort());
return serviceInterfaceInfo;
}
//Zookeeper服务发现
public ServiceInterfaceInfo selectOneInstance(String serviceName) {
String servicePath = "/com/leixiaoshuai/easyrpc/service/" + serviceName;
final List<String> childrenNodes = zkClient.getChildren(servicePath);
return Optional.ofNullable(childrenNodes)
.orElse(new ArrayList<>())
.stream()
.map(node -> {
try {
// 将服务信息经过 URL 解码后反序列化为对象
String serviceInstanceJson = URLDecoder.decode(node, "UTF-8");
return JSON.parseObject(serviceInstanceJson, ServiceInterfaceInfo.class);
} catch (UnsupportedEncodingException e) {
logger.error("Fail to decode", e);
}
return null;
}).filter(Objects::nonNull).findAny().get();
}