七 序列化
RPC框架中，离不开网络请求，服务A调用服务B的方法，要发送一个网络请求，服务B收到网络请求后，解析请求，然后把方法的执行结果返回给服务A。为了实现这些步骤，需要编写消息请求体、消息相应体、序列化与反序列化的相关逻辑。下面一起来看看怎么写吧～
import lombok.Data; import java.io.Serializable; /** * RPC请求体 * * @author zhongmingyi * @date 2021/12/12 1:01 下午 */ @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; /** * @author zhongmingyi * @date 2021/12/11 12:20 下午 */ 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; /** * RPC响应 * * @author zhongmingyi * @date 2021/12/12 4:38 下午 */ @Data public class RpcResponse implements Serializable { /** * 状态 */ private String status; /** * 返回结果 */ private Object value; /** * 异常 */ private Exception exception; }
那么对于网络传输，我们需要把对象序列化成byte数组，然后我们要操作对象的话，则需要把byte数组反序列化成对象。
1 定义序列化协议
写一个接口，来约定序列化的协议
/** * 序列化协议 * * @author zhongmingyi * @date 2021/12/14 4:03 下午 */ public interface MessageSerializationProtocol { byte[] marshal(Object object) throws Exception; <T> T unMarshal(byte[] bytes, Class<T> clazz) throws Exception; }
marshal：将对象序列化成byte[]
unMarshal：将byte[]反序列化成指定Class的对象
2 JDK序列化
定义 JdkMessageSerializationProtocol 类实现 MessageSerializationProtocol接口，JDK序列化的实现如下：
import com.zhongger.rpc.serialization.MessageSerializationProtocol; import java.io.*; /** * JDK序列化 * * @author zhongmingyi * @date 2021/12/14 4:13 下午 */ 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> T unMarshal(byte[] bytes, Class<T> 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; /** * Kryo序列化 * * @author zhongmingyi * @date 2021/12/14 5:32 下午 */ public class KryoMessageSerializationProtocol implements MessageSerializationProtocol { /** * Kryo是非线程安全的，需要ThreadLocal来防止出现线程安全问题 */ private final ThreadLocal<Kryo> 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> T unMarshal(byte[] bytes, Class<T> 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); } }
好了序列化协议就这样写好了，接下来，就要写比较复杂的网络通信咯～
