小新是也

小新是也

关注
发信
关注(28)粉丝(399)

从零开始实现简单 RPC 框架 2：扩展利器 SPI

RPC 框架有很多可扩展的地方，如：序列化类型、压缩类型、负载均衡类型、注册中心类型等等。
假设框架提供的注册中心只有zookeeper，但是使用者想用Eureka，修改框架以支持使用者的需求显然不是好的做法。
最好的做法就是留下扩展点，让使用者可以不需要修改框架，就能自己去实现扩展。
JDK 原生已经为我们提供了 SPI 机制，ccx-rpc 在此基础上，进行了性能优化和功能增强。
在讲解 ccx-rpc 的增强 SPI 之前，先来了解一下 JDK SPI 吧。

JDK SPI

下面我们来看一下 JDK SPI 是如何使用的。
我们先来定义一个序列化接口和 JSON、Protostuff 两种实现：

public interface Serializer {
    byte[] serialize(Object object);
}

public class JSONSerializer implements Serializer {
    @Override
    public byte[] serialize(Object object) {
        return JSONUtil.toJsonStr(object).getBytes();
    }
}

public class ProtostuffSerializer implements Serializer {
    private static final LinkedBuffer BUFFER = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
    @Override
    public byte[] serialize(Object object) {
        Schema schema = RuntimeSchema.getSchema(object.getClass());
        return ProtostuffIOUtil.toByteArray(object, schema, BUFFER);
    }
}

在 resources/META-INF/services 目录下添加一个 com.xxx.Serializer 的文件，这是 JDK SPI 的配置文件：

com.xxx.JSONSerializer
com.xxx.ProtostuffSerializer

如何使用 SPI 将实现类加载出来呢？

public static void main(String[] args) {
    ServiceLoader<Serializer> serviceLoader = ServiceLoader.load(Serializer.class);
    Iterator<Serializer> iterator = serviceLoader.iterator();
    while (iterator.hasNext()) {
        Serializer serializer= iterator.next();
        System.out.println(serializer.getClass().getName());
    }
}

输出如下:

com.xxx.JSONSerializer
com.xxx.ProtostuffSerializer

通过上面的例子，我们可以了解到 SPI 的简单用法。接下来，我们就来看增强版的 SPI 是如何实现的，又增强在哪里。

增强版 SPI

我们先来看看增强版 SPI 是如何使用的吧，还是拿序列化来举例。

  1. 定义接口，接口加上 @SPI 注解
@SPI
public interface Serializer {
    byte[] serialize(Object object);
}
  1. 实现类，这个代码跟上面的一模一样，就不重复贴代码了
  2. 配置文件
json=com.ccx.rpc.demo.client.spi.JSONSerializer
protostuff=com.ccx.rpc.demo.client.spi.ProtostuffSerializer
  1. 获取扩展类
    我们可以只实例化想要的实现类
public static void main(String[] args) {
    ExtensionLoader<Serializer> loader = ExtensionLoader.getLoader(Serializer.class);
    Serializer serializer = loader.getExtension("protostuff");
    System.out.println(serializer.getClass().getName());
}

上面是增强版 SPI 的基础用法，还是相当简单的。下面我们就要开始讲解代码实现了，准备好，要发车了。

懒惰加载

JDK SPI 在查找实现类的时候，需要遍历配置文件中定义的所有实现类，而这个过程会把所有实现类都实例化。一个接口如果有很多实现类，而我们只需要其中一个的时候，就会产生其他不必要的实现类。 例如 Dubbo 的序列化接口，实现类就有 fastjson、gson、hession2、jdk、kryo、protobuf 等等，通常我们只需要选择一种序列化方式。如果用 JDK SPI，那其他没用的序列化实现类都会实例化，实例化所有实现类明显是资源浪费！

ccx-rpc 的扩展加载器就对此进行了优化，只会对需要实例化的实现类进行实例化，也就是俗称的"懒惰加载"。

获取扩展类实例的实现如下：

public T getExtension(String name) {
    T extension = extensionsCache.get(name);
    if (extension == null) {
        synchronized (lock) {
            extension = extensionsCache.get(name);
            if (extension == null) {
                extension = createExtension(name);
                extensionsCache.put(name, extension);
            }
        }
    }
    return extension;
}

这是一个典型的 double-check 懒汉单例实现，当程序需要某个实现类的时候，才会去真正初始化它。

配置文件

配置文件采用的格式参考 dubbo，示例：

json=com.ccx.rpc.demo.client.spi.JSONSerializer
protostuff=com.ccx.rpc.demo.client.spi.ProtostuffSerializer

采用 key-value 的配置格式有个好处就是，要获取某个类型的扩展，可以直接使用名字来获取，可以大大提高可读性。

加载解析配置文件的代码也比较简单：

/**
 * 从资源文件中加载所有扩展类
 */
private Map<String, Class<?>> loadClassesFromResources() {
    // ... 省略非关键代码
    Enumeration<URL> resources = classLoader.getResources(fileName);
    while (resources.hasMoreElements()) {
        URL url = resources.nextElement();
        try (BufferedReader reader = new BufferedReader(url...) {
            // 开始读文件
            while (true) {
                String line = reader.readLine();
                parseLine(line, extensionClasses);
            }
        }
    }
}

 /**
  * 解析行，并且把解析到的类，放到 extensionClasses 中
  */
 private void parseLine(String line, Map<String, Class<?>> extensionClasses) {
     // 用等号将行分割开，kv[0]就是名字，kv[1]就是类名
     String[] kv = line.split("=");
     Class<?> clazz = ExtensionLoader.class.getClassLoader().loadClass(kv[1]);
     extensionClasses.put(kv[0], clazz);
 }

扩展类的创建

当获取扩展类不存在时，会加锁实例化扩展类。实例化的流程如下：

  1. 从配置文件中，加载该接口所有的实现类的 Class 对象，并放到缓存中。
  2. 根据要获取的扩展名字，找到对应的 Class 对象。
  3. 调用 clazz.newInstance() 实例化。(Class 需要有无参构造函数)
private T createExtension(String name) {
    // 获取当前类型所有扩展类
    Map<String, Class<?>> extensionClasses = getAllExtensionClasses();
    // 再根据名字找到对应的扩展类
    Class<?> clazz = extensionClasses.get(name);
    return (T) clazz.newInstance();
}

加载器缓存

加载器指的就是 ExtensionLoader<T>，为了减少对象的开销，ccx-rpc 屏蔽了加载器的构造函数，提供了一个静态方法来获取加载器。

/**
 * 扩展加载器实例缓存 {类型：加载器实例}
 */
private static final Map<Class<?>, ExtensionLoader<?>> extensionLoaderCache = new ConcurrentHashMap<>();

public static <S> ExtensionLoader<S> getLoader(Class<S> type) {
    // ... 忽略部分代码
    SPI annotation = type.getAnnotation(SPI.class);
    ExtensionLoader<?> extensionLoader = extensionLoaderCache.get(type);
    if (extensionLoader != null) {
        return (ExtensionLoader<S>) extensionLoader;
    }
    extensionLoader = new ExtensionLoader<>(type);
    extensionLoaderCache.putIfAbsent(type, extensionLoader);
    return (ExtensionLoader<S>) extensionLoader;
}

extensionLoaderCache 是一个 Map，缓存了各种类型的加载器。获取的时候先从缓存获取，缓存不存在则去实例化，然后放到缓存中。这是一个很常见的缓存技巧。

默认扩展

ccx-rpc 还提供了默认扩展的功能，接口在使用 @SPI 的时候可以指定一个默认的实现类名，例如 @SPI("netty")。
这样当获取扩展名留空没有配置的时候，就会直接获取默认扩展，减少了配置的量。

在获取扩展类的时候，会从 @SPI 中获取 value()，把默认扩展名缓存起来。

private static String defaultNameCache;

public static <S> ExtensionLoader<S> getLoader(Class<S> type) {
    // ... 省略
    SPI annotation = type.getAnnotation(SPI.class);
    defaultNameCache = annotation.value();
    // ... 省略
}

获取默认扩展的代码就很简单了，直接使用了 defaultNameCache 去获取扩展。

public T getDefaultExtension() {
    return getExtension(defaultNameCache);
}

适配扩展

获取扩展类的时候，需要输入扩展名，这样就需要先从配置里面读到响应的扩展名，才能根据扩展名获取扩展类。这个过程稍显麻烦，ccx-rpc 还提供了一种适配扩展，可以动态从 URL 中读取对应的配置并自动获取扩展类。
下面我们来看一下用法：

@SPI
public interface RegistryFactory {

    /**
     * 获取注册中心
     *
     * @param url 注册中心的配置，例如注册中心的地址。会自动根据协议获取注册中心实例
     * @return 如果协议类型跟注册中心匹配上了，返回对应的配置中心实例
     */
    @Adaptive("protocol")
    Registry getRegistry(URL url);
}

public static void main(String[] args) {
    // 获取适配扩展
    RegistryFactory zkRegistryFactory = ExtensionLoader.getLoader(RegistryFactory.class).getAdaptiveExtension();
    URL url = URLParser.toURL("zk://localhost:2181");
    // 适配扩展自动从 ur 中解析出扩展名，然后返回对应的扩展类
    Registry registry = zkRegistryFactory.getRegistry(url);
}

从实例代码，可以看到，有一个@Adaptive("protocol") 注解，方法中有 URL 参数。其逻辑就是，SPI 从传进来的 URL 的协议中字段中，获取到扩展名 zk。

下面我们来看看获取适配扩展的代码是怎么实现的吧。

public T getAdaptiveExtension() {
    InvocationHandler handler = new AdaptiveInvocationHandler<T>(type);
    return (T) Proxy.newProxyInstance(ExtensionLoader.class.getClassLoader(),
            new Class<?>[]{type}, handler);
}

适配扩展类其实是一个代理类，接下来来看看这个代理类 AdaptiveInvocationHandler：

public class AdaptiveInvocationHandler<T> implements InvocationHandler {

    private final Class<T> clazz;

    public AdaptiveInvocationHandler(Class<T> tClass) {
        clazz = tClass;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        if (args.length == 0) {
            return method.invoke(proxy, args);
        }
        // 找 URL 参数
        URL url = null;
        for (Object arg : args) {
            if (arg instanceof URL) {
                url = (URL) arg;
                break;
            }
        }
        // 找不到 URL 参数，直接执行方法
        if (url == null) {
            return method.invoke(proxy, args);
        }

        Adaptive adaptive = method.getAnnotation(Adaptive.class);
        // 如果不包含 @Adaptive，直接执行方法即可
        if (adaptive == null) {
            return method.invoke(proxy, args);
        }

        // 从 @Adaptive#value() 中拿到扩展名的 key
        String extendNameKey = adaptive.value();
        String extendName;
        // 如果这个 key 是协议，从协议拿。其他的就直接从 URL 参数拿
        if (URLKeyConst.PROTOCOL.equals(extendNameKey)) {
            extendName = url.getProtocol();
        } else {
            extendName = url.getParam(extendNameKey, method.getDeclaringClass() + "." + method.getName());
        }
        // 拿到扩展名之后，就直接从 ExtensionLoader 拿就行了
        ExtensionLoader<T> extensionLoader = ExtensionLoader.getLoader(clazz);
        T extension = extensionLoader.getExtension(extendName);
        return method.invoke(extension, args);
    }
}

从配置中获取扩展的代码注释都有，我们在梳理一下流程：

  1. 从方法参数中拿到 URL 参数，拿不到就直接执行方法
  2. 获取配置 Key。从 @Adaptive#value() 拿扩展名的配置 key，如果拿不到就直接执行方法
  3. 获取扩展名。判断配置 key 是不是协议，如果是就拿协议类型，否则拿 URL 后面的参数。
    例如 URL 是：zk://localhost:2181?type=eureka
    • 如果 @Adaptive("protocol")，那么扩展名就是协议类型：zk
    • 如果 @Adaptive("type")，那么扩展名就是type 参数：eureka
  4. 最后根据扩展名获取扩展 extensionLoader.getExtension(extendName)

总结

RPC 框架扩展很重要，SPI 是一个很好的机制。
JDK SPI 获取扩展的时候，会实例化所有的扩展，造成资源的浪费。
ccx-rpc 自己实现了一套增强版的 SPI，有如下特点：

  • 懒惰加载
  • key-value 结构的配置文件
  • 加载器缓存
  • 默认扩展
  • 适配扩展

ccx-rpc 的 SPI 机制参考 Dubbo SPI，在它的基础上进行了精简和修改，在此对 Dubbo 表示感谢。

08-15 22:12
Powered by LMLPHP ©2024 小新是也 0.034724