手把手教你写一个RPC框架（三）

手把手教你写一个RPC框架（三）

承接着上一节，手把手教你写一个RPC框架（二） 我们定义了两个注解、编写了的连接Zookeeper注册中心的逻辑，这一节，我们继续来完成RPC框架吧！

七 序列化

RPC框架中，离不开网络请求，服务A调用服务B的方法，要发送一个网络请求，服务B收到网络请求后，解析请求，然后把方法的执行结果返回给服务A。为了实现这些步骤，需要编写消息请求体、消息相应体、序列化与反序列化的相关逻辑。下面一起来看看怎么写吧～

import lombok.Data;
import java.io.Serializable;


@Data
public class RpcRequest implements Serializable {
    
    private String serviceName;

    
    private String method;

    
    private Class[] methodParameterTypes;

    
    private Object[] methodParameters;

}

上面这四个属性是一个RPC请求中必不可少的

• 所请求的服务接口名
• 所请求的服务接口中，具体的方法名
• 所请求的方法的参数类型
• 所请求的方法的参数

有了这些属性，我们才能从注册中心中找到对应的服务。当然有小伙伴肯定会问，RpcRequest中并没有所请求服务的IP地址和端口号，该怎么找到对应的地址啊？

其实在上一节，在注册中心编写的代码中，就有写到注册的逻辑：

package com.zhongger.rpc.register.impl;

import com.alibaba.fastjson.JSON;
import com.zhongger.rpc.entity.ServerNode;
import com.zhongger.rpc.register.RpcServiceRegister;
import org.apache.curator.framework.Curatorframework;
import org.apache.curator.framework.CuratorframeworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URLEncoder;


public class ZookeeperRpcServiceRegister implements RpcServiceRegister {
 
       @Override
       public void register(ServiceNode serviceNode) throws Exception {
        logger.info("register server node info is {}", serviceNode);
        String uri = JSON.toJSONString(serviceNode);
        uri = URLEncoder.encode(uri, "UTF-8");

        String servicePath = "/com/zhongger/rpc/" + serviceNode.getServiceName() + "/service";
        // 创建永久节点
        if (zookeeperClient.checkExists().forPath(servicePath) == null) {
            logger.info("service path {} not exist create persistent node ", servicePath);
            zookeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(servicePath);
        }

        //创建临时节点
        String uriPath = servicePath + "/" + uri;
        logger.info("uri path is {}", uriPath);
        if (zookeeperClient.checkExists().forPath(uriPath) != null) {
            zookeeperClient.delete().forPath(uriPath);
        }
        zookeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(uriPath);
    }

}

其中ServerNode实体类就包含了IP地址、端口号、服务名称的信息，比如我有个接口HelloService，IP地址127.0.0.1，端口8888：注册到Zookeeper中的结构就是这样的，永久节点是：/com/zhongger/rpc/HelloService/service 临时节点则是：/com/zhongger/rpc/HelloService/service/{ServiceNode的JSON结构}，所以RpcRequest其实就不需要Ip地址和port这些属性了。另外，寻找对应方法的时候，只需要遍历/com/zhongger/rpc/HelloService/service下的所有子节点，然后通过负载均衡来选择一个Ip地址+Port来调用对应的方法就行啦，后面会详细介绍这里的实现。

写完了请求，那么来编写响应，响应就比较简单了，就是把方法的执行结果封装一下：

package com.zhongger.rpc.entity;

import lombok.Data;

import java.io.Serializable;



@Data
public class RpcResponse implements Serializable {
    
    private String status;
    
    private Object value;
    
    private Exception exception;
}

那么对于网络传输，我们需要把对象序列化成byte数组，然后我们要 *** 作对象的话，则需要把byte数组反序列化成对象。

1 定义序列化协议

写一个接口，来约定序列化的协议

public interface MessageSerializationProtocol {
    byte[] marshal(Object object) throws Exception;

     T unMarshal(byte[] bytes, Class clazz) throws Exception;
}

• marshal：将对象序列化成byte[]
• unMarshal：将byte[]反序列化成指定Class的对象

2 JDK序列化

定义 JdkMessageSerializationProtocol 类实现 MessageSerializationProtocol接口，JDK序列化的实现如下：

import com.zhongger.rpc.serialization.MessageSerializationProtocol;

import java.io.*;


public class JdkMessageSerializationProtocol implements MessageSerializationProtocol {
    @Override
    public byte[] marshal(Object object) throws Exception {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
        objectOutputStream.writeObject(object);
        byte[] result = byteArrayOutputStream.toByteArray();
        objectOutputStream.close();
        byteArrayOutputStream.close();
        return result;
    }

    @Override
    public  T unMarshal(byte[] bytes, Class clazz) throws Exception {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);
        T result = clazz.cast(objectInputStream.readObject());
        objectInputStream.close();
        byteArrayInputStream.close();
        return result;
    }

}

JDK序列化的方式效率是比较低的，于是我用了比较流行的Kryo序列化框架又实现了一套序列化协议

3 Kryo序列化

使用Kryo要注意：Kryo是非线程安全的，需要ThreadLocal来防止出现线程安全问题

package com.zhongger.rpc.serialization.impl;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.zhongger.rpc.entity.RpcRequest;
import com.zhongger.rpc.entity.RpcResponse;
import com.zhongger.rpc.serialization.MessageSerializationProtocol;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;



public class KryoMessageSerializationProtocol implements MessageSerializationProtocol {

    
    private final ThreadLocal kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(RpcRequest.class);
        kryo.register(RpcResponse.class);
        return kryo;
    });

    @Override
    public byte[] marshal(Object object) throws Exception {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);
        Kryo kryo = kryoThreadLocal.get();
        kryo.writeObject(output, object);
        kryoThreadLocal.remove();
        return output.toBytes();
    }

    @Override
    public  T unMarshal(byte[] bytes, Class clazz) throws Exception {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        Input input = new Input(byteArrayInputStream);
        Kryo kryo = kryoThreadLocal.get();
        Object o = kryo.readObject(input, clazz);
        kryoThreadLocal.remove();
        return clazz.cast(o);
    }
}

好了序列化协议就这样写好了，接下来，就要写比较复杂的网络通信咯～