序列化器
序列化接口
/*** * 序列化器接口*/
public interface Serializer {/*** 序列化** @param object* @param <T>* @return* @throws IOException*/<T> byte[] serialize(T object) throws IOException;/*** 反序列化** @param bytes* @param tClass* @param <T>* @return* @throws IOException*/<T> T deserialize(byte[] bytes, Class<T> tClass) throws IOException;}
序列化器实现类
jdk序列化器
/*** JDK 序列化器*/
public class JdkSerializer implements Serializer {/*** 序列化** @param object 序列化对象* @param <T> 泛型* @return 字节数组*/@Overridepublic <T> byte[] serialize(T object) throws IOException {ByteArrayOutputStream outputStream = new ByteArrayOutputStream();ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);objectOutputStream.writeObject(object);objectOutputStream.close();return outputStream.toByteArray();}/*** 反序列化** @param bytes 字节数组* @param type class对象* @param <T> 泛型* @return 对象*/@Overridepublic <T> T deserialize(byte[] bytes, Class<T> type) throws IOException {ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);try (ObjectInputStream objectInputStream = new ObjectInputStream(inputStream)) {return type.cast(objectInputStream.readObject());} catch (ClassNotFoundException e) {throw new RuntimeException(e);}}
}
JSON序列化器
统一封装RPC请求响应
/*** RPC 请求*/
public class RpcRequest implements Serializable {/*** 服务名称*/private String serviceName;/*** 方法名称*/private String methodName;/*** 服务版本*/private String serviceVersion = RpcConstants.DEFAULT_SERVICE_VERSION;/*** 参数类型列表*/private Class<?>[] parameterTypes;/*** 参数列表*/private Object[] args;// getter and setter...
}/*** RPC 响应*/
public class RpcResponse implements Serializable {/*** 响应数据*/private Object data;/*** 响应数据类型（预留）*/private Class<?> dataType;/*** 响应信息*/private String message;/*** 异常信息*/private Exception exception;// getter and setter...
}
/*** json序列化器*/
public class JsonSerializer implements Serializer {private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();@Overridepublic <T> byte[] serialize(T object) throws IOException {return OBJECT_MAPPER.writeValueAsBytes(object);}@Overridepublic <T> T deserialize(byte[] bytes, Class<T> classType) throws IOException {T obj = OBJECT_MAPPER.readValue(bytes, classType);if (obj instanceof RpcRequest) {return handleRequest((RpcRequest) obj, classType);}if (obj instanceof RpcResponse) {return handleResponse((RpcResponse) obj, classType);}return obj;}/*** 由于 Object 的原始对象会被擦除，导致反序列化时会被作为 LinkedHashMap 无法转换成原始对象，做特殊处理** @param rpcRequest rpc 请求* @param type 类型* @return {@link T}* @throws IOException IO异常*/private <T> T handleRequest(RpcRequest rpcRequest, Class<T> type) throws IOException {Class<?>[] parameterTypes = rpcRequest.getParameterTypes();Object[] args = rpcRequest.getArgs();// 循环处理每个参数的类型for (int i = 0; i < parameterTypes.length; i++) {Class<?> clazz = parameterTypes[i];// 如果类型不同，则重新处理一下类型if (!clazz.isAssignableFrom(args[i].getClass())) {byte[] argBytes = OBJECT_MAPPER.writeValueAsBytes(args[i]);args[i] = OBJECT_MAPPER.readValue(argBytes, clazz);}}return type.cast(rpcRequest);}/*** 由于 Object 的原始对象会被擦除，导致反序列化时会被作为 LinkedHashMap 无法转换成原始对象，因此这里做了特殊处理** @param rpcResponse rpc 响应* @param type 类型* @return {@link T}* @throws IOException IO异常*/private <T> T handleResponse(RpcResponse rpcResponse, Class<T> type) throws IOException {// 处理响应数据byte[] dataBytes = OBJECT_MAPPER.writeValueAsBytes(rpcResponse.getData());rpcResponse.setData(OBJECT_MAPPER.readValue(dataBytes, rpcResponse.getDataType()));return type.cast(rpcResponse);}
}
Hessian序列化器
/*** Hessian序列化器*/
public class HessianSerializer implements Serializer {/*** 序列化对象为byte数组** @param obj 需要序列化的对象* @return 序列化后的byte数组*/public <T> byte[] serialize(T obj) throws IOException {ByteArrayOutputStream bao = new ByteArrayOutputStream();HessianOutput ho = new HessianOutput(bao);ho.writeObject(obj);ho.flush();return bao.toByteArray();}/*** 反序列化byte数组为对象** @param bytes 需要反序列化的byte数组* @param clazz 反序列化对象的类类型* @param <T> 反序列化对象的泛型类型* @return 反序列化后的对象*/public <T> T deserialize(byte[] bytes, Class<T> clazz) throws IOException {ByteArrayInputStream ba = new ByteArrayInputStream(bytes);HessianInput hi = new HessianInput(ba);return clazz.cast(hi.readObject());}
}
Kryo序列化器
/*** Kryo序列化器*/
public class KryoSerializer implements Serializer {/*** kryo 线程不安全，使用 ThreadLocal 保证每个线程只有一个 Kryo*/private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {Kryo kryo = new Kryo();// 设置动态动态序列化和反序列化类，不提前注册所有类（可能有安全问题）kryo.setRegistrationRequired(false);return kryo;});/*** 序列化对象为byte数组** @param obj 需要序列化的对象* @return 序列化后的byte数组*/public <T> byte[] serialize(T obj) {Kryo kryo = kryoThreadLocal.get();ByteArrayOutputStream bao = new ByteArrayOutputStream();Output output = new Output(bao);kryo.writeObject(output, obj);output.close();return bao.toByteArray();}/*** 反序列化byte数组为对象** @param bytes 需要反序列化的byte数组* @param clazz 反序列化对象的类类型* @param <T> 反序列化对象的泛型类型* @return 反序列化后的对象*/public <T> T deserialize(byte[] bytes, Class<T> clazz) {Kryo kryo = kryoThreadLocal.get();ByteArrayInputStream ba = new ByteArrayInputStream(bytes);Input input = new Input(ba);T object = kryo.readObject(input, clazz);input.close();return object;}
}
自定义SPI机制
SPI加载器
系统级SPI和用户自定义SPI，且用户SPI优先级更高
/*** SPI 加载器（支持键值对映射）*/
public class SpiLoader {private static final Logger logger = LoggerFactory.getLogger(SpiLoader.class);/*** 存储已加载的类：接口名 =>（key => 实现类）*/private static final Map<String, Map<String, Class<?>>> loaderMap = new ConcurrentHashMap<>();/*** 对象实例缓存（避免重复 new），类路径 => 对象实例，单例模式*/private static final Map<String, Object> instanceCache = new ConcurrentHashMap<>();/*** 系统默认 SPI 目录*/private static final String RPC_DEFAULT_SPI_DIR = "META-INF/rpc/default/";/*** 用户自定义 SPI 目录*/private static final String RPC_CUSTOM_SPI_DIR = "META-INF/rpc/custom/";/*** 扫描路径*/private static final String[] SCAN_DIRS = new String[]{RPC_DEFAULT_SPI_DIR, RPC_CUSTOM_SPI_DIR};/*** 动态加载的类列表*/private static final List<Class<?>> LOAD_CLASS_LIST = Arrays.asList(Serializer.class);/*** 加载所有类型*/public static void loadAll() {logger.info("加载所有 SPI");for (Class<?> aClass : LOAD_CLASS_LIST) {load(aClass);}}/*** 获取某个接口的实例** @param tClass* @param key* @param <T>* @return*/public static <T> T getInstance(Class<?> tClass, String key) {String tClassName = tClass.getName();Map<String, Class<?>> keyClassMap = loaderMap.get(tClassName);if (keyClassMap == null) {throw new RuntimeException(String.format("SpiLoader 未加载 %s 类型", tClassName));}if (!keyClassMap.containsKey(key)) {throw new RuntimeException(String.format("SpiLoader 的 %s 不存在 key=%s 的类型", tClassName, key));}// 获取到要加载的实现类型Class<?> implClass = keyClassMap.get(key);// 从实例缓存中加载指定类型的实例String implClassName = implClass.getName();if (!instanceCache.containsKey(implClassName)) {try {instanceCache.put(implClassName, implClass.newInstance());} catch (InstantiationException | IllegalAccessException e) {String errorMsg = String.format("%s 类实例化失败", implClassName);throw new RuntimeException(errorMsg, e);}}return (T) instanceCache.get(implClassName);}/*** 加载某个类型** @param loadClass 加载类型* @throws IOException*/public static Map<String, Class<?>> load(Class<?> loadClass) {logger.info("加载类型为 {} 的 SPI", loadClass.getName());// 扫描路径，用户自定义的 SPI 优先级高于系统 SPI【map中相同key覆盖保证优先级】Map<String, Class<?>> keyClassMap = new HashMap<>();for (String scanDir : SCAN_DIRS) {List<URL> resources = ResourceUtil.getResources(scanDir + loadClass.getName());// 读取每个资源文件for (URL resource : resources) {try {InputStreamReader inputStreamReader = new InputStreamReader(resource.openStream());BufferedReader bufferedReader = new BufferedReader(inputStreamReader);String line;while ((line = bufferedReader.readLine()) != null) {String[] strArray = line.split("=");if (strArray.length > 1) {String key = strArray[0];String className = strArray[1];keyClassMap.put(key, Class.forName(className));}}} catch (Exception e) {logger.error("spi resource load error", e);}}}loaderMap.put(loadClass.getName(), keyClassMap);return keyClassMap;}
}
SPI文件名为类全包名，内容为键值对，如下：
