本文讲述了flink Java教程——Java SPI 机制在 Flink 中的应用！具有很好的参考价值，希望对大家有所帮助。一起跟随六星小编过来看看吧，具体如下:

我们在使用 Flink SQL 的时候是否有过这样的疑问? Flink 提供了各种各样的 connector 我们只需要在 DML 里面定义即可运行,那它是怎么找到要执行的代码呢? 它是怎么知道代码对应关系的呢? 其实 Flink 是通过 Java 的 SPI(并不是Flink发明创造的) 机制来实现的,下面就来深入源码分析一下其实现过程.

什么是 SPI ?

SPI 全称（Service Provide Interface），在 JAVA 中是一个比较重要的概念，在框架设计中被广泛使用。在框架设计中，要遵循的原则是对扩展开放，对修改关闭，保证框架实现对于使用者来说是黑盒。因为框架不可能做好所有的事情，只能把共性的部分抽离出来进行流程化，然后留下一些扩展点让使用者去实现，这样不同的扩展就不用修改源代码或者对框架进行定制。也就是我们经常说的面向接口编程。

在 JDK6 里面引进的一个新的特性 ServiceLoader，从官方的文档来说，它主要是用来装载一系列的 service provider。而且ServiceLoader 可以通过 service provider 的配置文件来装载指定的 service provider。当服务的提供者，提供了服务接口的一种实现之后，我们只需要在 jar 包的 META-INF/services/ 目录里同时创建一个以服务接口命名的文件。该文件里就是实现该服务接口的具体实现类。而当外部程序装配这个模块的时候，就能通过该 jar 包 META-INF/services/ 里的配置文件找到具体的实现类名，并装载实例化，完成模块的注入。综上所述，SPI 机制实际上就是 "基于接口的编程+策略模式+配置文件" 组合实现的一种动态加载机制，在 JDK 中提供了工具类：java.util.ServiceLoader 来实现服务查找。

实现 SPI  机制，要遵循下面的一些规范：

服务提供者提供了接口的具体实现后，需要在资源文件夹中创建 META-INF/services 文件夹，并且新建一个以全类名为名字的文本文件，文件内容为实现类的全名（如下面图中的红框）；

接口实现类必须在工程的 classpath 下，也就是 maven 中需要加入依赖或者 jar 包引用到工程里.

其实 Java 里使用 SPI 还是比较多的,比如我们常用的 JDBC 连接 Mysql 就是用的 SPI 机制来实现的连接逻辑,下面来看一个简单的 Demo.

接口服务提供

public interface Person {

    void eat();

}

实现类

public class Flink implements Person {

    @Override

    public void eat() {

        System.out.println(this.getClass().getSimpleName() + " 执行 eat 方法");

    }

}

public class JasonLee implements Person {

    @Override

    public void eat() {

        System.out.println(this.getClass().getSimpleName() + " 执行 eat 方法");

    }

}

public class Spark implements Person {

    @Override

    public void eat() {

        System.out.println(this.getClass().getSimpleName() + " 执行 eat 方法");

    }

}

META-INF/services 配置文件

在 source 下面的  META-INF/services 文件夹下面创建 spi.Person 文件,文件的内容是上面实现类的全名(包名.类名)

测试类

public static void main(String[] args) {

    ServiceLoader<Person> load = ServiceLoader.load(Person.class);

    Iterator<Person> iterator = load.iterator();

    while (iterator.hasNext()) {

        Person next = iterator.next();

        next.eat();

    }

}

执行打印的结果是:

Flink 执行 eat 方法

JasonLee 执行 eat 方法

Spark 执行 eat 方法

可以看到 3 个实现类都被执行了.

然后来分析一下 ServiceLoader#load 方法的源码:

public static <S> ServiceLoader<S> load(Class<S> service) {

    ClassLoader cl = Thread.currentThread().getContextClassLoader();

    return ServiceLoader.load(service, cl);

}

首先会获取当前线程的类加载器,然后调用另一个重载的 load 方法.

public static <S> ServiceLoader<S> load(Class<S> service,

                                        ClassLoader loader)

{

    return new ServiceLoader<>(service, loader);

}

这个 load 方法会调用 ServiceLoader 的构造方法进行变量初始化.

private ServiceLoader(Class<S> svc, ClassLoader cl) {

    service = Objects.requireNonNull(svc, "Service interface cannot be null");

    loader = (cl == null) ? ClassLoader.getSystemClassLoader() : cl;

    acc = (System.getSecurityManager() != null) ? AccessController.getContext() : null;

    reload();

}

构造方法里面主要是完成了对 service,loader,acc 变量的赋值工作,然后调用 reload 方法

那么 Flink 是如何保证正确的 TableFactory 实现类被加载的呢？直接来看 TableFactoryService#findSingleInternal 方法的源码

private static <T extends TableFactory> T findSingleInternal(

        Class<T> factoryClass,

        Map<String, String> properties,

        Optional<ClassLoader> classLoader) {

    // 加载所有的实现类

    List<TableFactory> tableFactories = discoverFactories(classLoader);

    // 过滤出满足条件的

    List<T> filtered = filter(tableFactories, factoryClass, properties);

    if (filtered.size() > 1) {

        throw new AmbiguousTableFactoryException(

                filtered, factoryClass, tableFactories, properties);

    } else {

        return filtered.get(0);

    }

}

其中 discoverFactories 方法用来发现并加载 Table 的服务提供类，filter 方法则是用来筛选出满足条件的 TableFactory 的实现类。前者最终调用了 ServiceLoader 的相关方法，如下：

private static List<TableFactory> discoverFactories(Optional<ClassLoader> classLoader) {

    try {

        List<TableFactory> result = new LinkedList<>();

        ClassLoader cl = classLoader.orElse(Thread.currentThread().getContextClassLoader());

        ServiceLoader.load(TableFactory.class, cl).iterator().forEachRemaining(result::add);

        return result;

    } catch (ServiceConfigurationError e) {

        LOG.error("Could not load service provider for table factories.", e);

        throw new TableException("Could not load service provider for table factories.", e);

    }

}

可以看到 Java 的 SPI 机制就是在这里用的,查找并加载 TableFactory 所有的实现类,然后保存在 List 里面.然后在过滤中满足条件的一个.