动手实现RPC
最近利用业于时间实现一个了一个简简单单的 RPC demo，在这里过程中，遇到了几个问题，也收获一些东西,分享一下这个过程.
前期准备
RPC 是我们在企业开发中比较常见的，同时也是比较熟悉的，但是对于开发一个 RPC 来说，我们需要掌握一些基本的理论知识.
1、RPC 是网络通信(进程之间的通信)，所以相比单进程的通信来的慢
2、网络传输的是 字节 ，而不是 字符，或者其他的传输媒介.
3、了解or掌握了 动态代理的运用。
本质是传输信息，将 我(消费端) 所知的信息通过通信告诉你 (服务端) ，然后你给我返回最终的信息.
动手实现
有了上述的前期准备之后，先剖析一下我们是如何使用 RPC ，服务提供端需要将服务暴露出去，服务消费端需要接入需要的服务。这个过程如下所示，我的讲述也围绕这个过程展开.
服务端暴露服务 --> 消费端引用服务 --> 动态代理 --> 序列化 --> 网络请求 ---> 服务端处理 --> 序列化返回 --> 消费端返回结果.
服务暴露
服务暴露就是将服务发布到(zookeeper)上，提供给其他的消费者访问，当消费者拿到这个信息之后，就可以连接服务端，并发起请求。
暴露就是需要将该服务的基本信息发布出去，这里列举的主要信息有 服务机器host 服务监听端口 服务接口 服务序列化方式 服务权重 服务版本 等
代码如下
1、服务暴露的信息
public class Provider implements Serializable, Cloneable {
private String serviceName;
private String host;
private Integer port;
private String version;
private Integer weight;
private String serialization;
get and set
}
2、暴露服务的过程，这里以 zookeeper 为例，主要是将服务以节点的形式添加到 zk 上
@Override
public void registerService(List<Provider> providerList) {
assert zkClient != null;
providerList.parallelStream().forEach(provider -> {
String host = provider.getHost();
Integer port = provider.getPort();
String serviceName = provider.getServiceName();
String version = provider.getVersion();
Integer weight = provider.getWeight();
String serverPath = root_path + "/" + serviceName + root_provider;
if (!zkClient.exists(serverPath)) {
zkClient.createPersistent(serverPath, true);
}
String finalInfo = host + split + port + split + serviceName + split + version + split + weight + split + provider.getSerialization();
String path = serverPath + "/" + finalInfo;
if (!zkClient.exists(path)) {
log.info("注册服务:{}到ZooKeeper", serverPath);
zkClient.createEphemeral(path);
} else {
log.warn("服务:{}已被注册", serverPath);
}
});
}
然后可以通过 zkCli 命令查看zk上服务的状况.
[zk: localhost:2181(CONNECTED) 0] ls /fuck/top.huzhurong.fuck.UserService/provider
[]
消费端引用服务和动态代理
消费端引用服务端方式大多数还是通过 Spring 的自定义标签引入.使用方式如下.
<fuck:reference id="test" interface="top.huzhurong.fuck.UserService" version="0.0.1"/>
这样就引用了 top.huzhurong.fuck.UserService 这个版本为 0.0.1 的服务. 在使用Spring的时候，我们需要通过使用动态代理去走网络，调用服务上的接口，那么就不可能想普通的 bean
一样去配置，一般说来都是通过 FactoryBean 来进行配置，因为在 getObject 中可以进行 bean的定制化(大部分的框架也是FactoryBean引入的)。
FactoryBean的使用方式如下
public class ProxyBean implements FactoryBean, InitializingBean {
private Class name;
private Object object;
@Override
public Object getObject() {
return object;
}
@Override
public Class<?> getObjectType() {
return name;
}
@Override
public void afterPropertiesSet() {
this.build();
}
private void build() {
//动态代理产生一个代理bean
this.object = Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[]{this.name}, (proxy, method, args) -> {
//在invoke中定制我们的服务，可以走tcp/http等等
if (method.getName().equalsIgnoreCase("name")) {
return "调用name方法";
}
if (method.getName().equalsIgnoreCase("toString")) {
return "调用toString方法";
}
return "111";
});
}
}
这里只是一个简单的说明服务引用的过程，具体的过程如下
- 注册消费节点到zookeeper
- 获取服务者列表
- 订阅该服务接口
- 搭建动态代理
篇幅有限，我就介绍下搭建动态代理，其他三个还挺简单的。
Object object = Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[]{Class.forName(this.interfaceName)}
, new FuckRpcInvocationHandler(this));
class FuckRpcInvocationHandler implements InvocationHandler {
//字段
//构造方法
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//获取服务列表
List<Provider> all = ProviderSet.getAll(this.className);
//软负载均衡
Provider provider = loadBalance.getProvider(all);
String info = provider.buildIfno();
SocketChannel channel = ChannelMap.get(info);
if (channel == null) {
//建立TCP连接
Client client = new NettyClient(provider, this.serialization);
client.connect(provider.getHost(), provider.getPort());
channel = ChannelMap.get(info);
}
SocketChannel finalChannel = channel;
Future<Response> submit = TempResultSet.executorService.submit(() -> {
//请求写入tcp通道
finalChannel.writeAndFlush(request);
//这里可以改造成 CountDownLatch,可以比 ;; 循环要好
for (; ; ) {
Response response = TempResultSet.get(request.getRequestId());
if (response != null) {
return response;
}
}
});
Response response = submit.get(this.timeout, TimeUnit.SECONDS);
}
}
动态代理做的主要是，获取服务列表，软件负载处理，建立tcp连接，通信这几步.
动态代理仅仅是实际调用的时候才会进入invoke方法，实例化不进入，每一次代理调用都会进入
invoke
序列化和网络请求
序列化网上都文章很多，我也只是支持了 protostuff 和 jdk 序列化方式. 本身很难，都是封装之后还是ok的，重点介绍一下网络传输的处理
当解决完序列化之后，就是网络传输了，都知道网络传输的是字节，但是怎么去使用网络进行透明传输我们都很头痛，而netty在使用上降低了我们的入门难度，其简单的api可以快速的进行上手，如果你还没有通过，可以看看netty的example就可以动手写了。
在网络传输部分的处理中，一个是协议的处理，一个是如何将处理完的请求结果赋值给正确的请求对象.
1、协议的处理也非常简单，就是一个常规的length + data ， 作为自定义协议，这种处理可以让我们快速的处理而不会纠结于协议的正确性
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> list) {
if (byteBuf.readableBytes() <= HEAD_LENGTH) {
return;
}
byteBuf.markReaderIndex();//标记位置
int dataLength = byteBuf.readInt();
if (byteBuf.readableBytes() < dataLength) {
byteBuf.resetReaderIndex();//可读取的数据不够
return;
}
byte[] dataArray = new byte[dataLength];
byteBuf.readBytes(dataArray);
Request request = serialization.deSerialize(dataArray, Request.class);
if (log.isDebugEnabled()) {
log.debug("接受到消费者请求:{},请求内容:{}", ctx.channel().toString(), request);
}
list.add(request);
}
2、如何将处理结果返回给正确的请求对象，我们都知道在执行请求的时候都需要一个 RI --> requestId 去标示这个请求，在我们的rpc当中也是如此，确保请求和响应的是一家人
public class Request implements Serializable {
private String requestId;//请求标示
private String serviceName;//服务名称
private String methodName;//方法名，Method不能被序列化
private Class<?>[] parameters;//参数类型，用于获取对于的执行方法
private Object[] args;//实际参数
}
3、服务处理，可以选择新建立线程池而不是直接使用 netty 的work io 线程池。
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Serializable serializable) {
if (serializable instanceof Request) {
Request request = (Request) serializable;
responseTask.execute(new ResponseTask(request, channelHandlerContext, this.applicationContext));
}
}
@Override
public void run() {
String serviceName = request.getServiceName();
String methodName = request.getMethodName();
Class<?>[] parameters = request.getParameters();
Object[] args = request.getArgs();
Response response = new Response();
response.setRequestId(request.getRequestId());
response.setSuccess(false);
try {
//获取服务
Object service = ServiceCache.getService(serviceName);
if (service == null) {
Class<?> aClass = ClassUtils.forName(serviceName, ClassUtils.getDefaultClassLoader());
service = applicationContext.getBean(aClass);
ServiceCache.put(serviceName, service);
}
//获取方法
Method method = service.getClass().getDeclaredMethod(methodName, parameters);
Object invoke = method.invoke(service, args);
System.out.println("invoke:" + invoke);
response.setSuccess(true);
response.setObject(invoke);
} catch (ClassNotFoundException | IllegalAccessException e) {
response.setException(e);
} catch (NoSuchMethodException e) {
e.printStackTrace();
response.setException(e);
} catch (InvocationTargetException e) {
e.printStackTrace();
response.setException(e.getTargetException());
}
//写入channel中
channelHandlerContext.writeAndFlush(response);
}
消费端返回结果
当服务端写入数据的时候，如果网络通畅，基本上一下就到了客户端，我们的处理也很简单，就是放到一个Map中，而消费线程一直在map中找这个数据
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, Serializable serializable) {
if (serializable instanceof Response) {
Response response = (Response) serializable;
ResponseFuture responseFuture = TempResultSet.getResponseFuture(response.getRequestId());
if (responseFuture == null) {
return;
}
if (response.getAsync()) {
CompletableFuture<Object> future = responseFuture.getFuture();
if (future != null) {
if (response.getObject() != null) {
future.complete(response.getObject());
} else {
assert response.getException() != null;
future.completeExceptionally(response.getException());
}
}
} else {
//解除CountDownLatch
responseFuture.putResponse(response);
}
}
}
到这里一个简单的rpc调用就可以起来了。
小结
一个简单的 rpc 就新鲜出炉了，但是其实还有很多可以改造的点，例如拦截(责任链+SPI)，例如观察者模式的运用，不过也是从中学习到了一个RPC基本的使用方法，还需要深入到了解线程池使用，这里的使用也是很粗制滥造的。不过能在几天之内写完还是很开心，项目在这里 ----> fuck-rpc 的简单实现，你也可以试试哦
