DataX源码分析-插件机制

系列文章目录

一、DataX详解和架构介绍
二、DataX源码分析 JobContainer
三、DataX源码分析 TaskGroupContainer
四、DataX源码分析 TaskExecutor
五、DataX源码分析 reader
六、DataX源码分析 writer
七、DataX源码分析 Channel
八、DataX源码分析-插件机制

---

---

前言

DataX的插件机制是其核心特性之一，它使得DataX能够灵活地适应各种不同的数据源的数据同步。这一机制主要基于插件开发框架，该框架主要包括Reader插件、Transformer插件、Writer插件。

DataX的插件机制还采用了框架+插件的架构。框架负责连接Reader和Writer插件，作为两者的数据传输通道，并处理缓冲、流控、并发、数据转换等核心技术问题。这种架构使得插件只需关心数据的读取或写入本身，而同步的共性问题则由框架来处理。

此外，DataX的插件机制还具有良好的扩展性和可维护性。开发者可以根据需要开发新的Reader或Writer插件来支持新的数据源类型，而无需修改DataX的核心框架代码。这种插件化的设计使得DataX能够适应不断变化的业务需求和技术环境。

在插件的加载和初始化方面，DataX使用了类似Java SPI（Service Provider Interface）的机制。它会在指定的插件目录中查找并加载插件，然后将其注册到插件注册中心。这样，当需要使用某个插件时，就可以从注册中心中获取其实例，并进行相应的操作。

总的来说，DataX的插件机制是一种非常灵活和可扩展的设计，它使得DataX能够适应各种不同的数据源和数据存储需求，同时也为开发者提供了丰富的扩展和定制化的可能性。

---

一、插件分类

按照功能分：
reader， 读插件，例如mysqlReader，从mysql读取数据
writer， 写插件。例如mysqlWriter，给mysql写入数据；
transformer， 中间结果转换，例如SubstrTransformer用于字符截取；
按照运行类型分：
Job级别的插件
Task级别的插件

插件目录结构

datax\plugin下分2个reader和writer目录，下面以mysql为例

plugin.json内容：

{"name": "mysqlreader","class": "xxx.plugin.reader.mysqlreader.MysqlReader","description": "useScene: prod. mechanism: Jdbc connection using the database, execute select sql, retrieve data from the ResultSet. warn: The more you know about the database, the less problems you encounter.","developer": "xx"
}

插件加载原理

• DataX进程启动入口为com.alibaba.datax.core.Engineengine.entry()

    public static void entry(final String[] args) throws Throwable {Options options = new Options();options.addOption("job", true, "Job config.");options.addOption("jobid", true, "Job unique id.");options.addOption("mode", true, "Job runtime mode.");BasicParser parser = new BasicParser();CommandLine cl = parser.parse(options, args);String jobPath = cl.getOptionValue("job");// 如果用户没有明确指定jobid, 则 datax.py 会指定 jobid 默认值为-1String jobIdString = cl.getOptionValue("jobid");RUNTIME_MODE = cl.getOptionValue("mode");Configuration configuration = ConfigParser.parse(jobPath);}

• 读取并解析插件配置
  ConfigParser.parse(final String jobPath)传入job路径，该方法组装解析，最后返回一个Configuration对象，Configuration里解析出了reader，writer，handler等插件名称；提取完插件名称后，会去reader目录和writer目录，寻找插件的位置。
• 动态加载插件
  插件的加载都是通过自定义类加载器JarLoader动态加载,提供插件相关Jar隔离的加载机制。插件的加载接口由LoadUtil类负责，当要加载一个插件时，需要实例化一个JarLoader，然后切换thread class loader之后，才加载插件。这个主要由ClassLoaderSwapper实现。
• JarLoader类
  JarLoader 负责加载指定路径下的插件 JAR 文件。它会检查 JAR 文件的合法性、有效性以及是否包含必要的插件实现类。继承自URLClassLoader提供Jar隔离的加载机制，会把传入的路径、及其子路径、以及路径中的jar文件加入到class path。

/*** 提供Jar隔离的加载机制，会把传入的路径、及其子路径、以及路径中的jar文件加入到class path。*/
public class JarLoader extends URLClassLoader{public JarLoader(String[] paths) {this(paths, JarLoader.class.getClassLoader());}public JarLoader(String[] paths, ClassLoader parent) {super(getURLs(paths), parent);}private static URL[] getURLs(String[] paths) {Validate.isTrue(null != paths && 0 != paths.length,"jar包路径不能为空.");List<String> dirs = new ArrayList<String>();for (String path : paths) {dirs.add(path);JarLoader.collectDirs(path, dirs);}List<URL> urls = new ArrayList<URL>();for (String path : dirs) {urls.addAll(doGetURLs(path));}return urls.toArray(new URL[0]);}private static void collectDirs(String path, List<String> collector) {if (null == path || StringUtils.isBlank(path)) {return;}File current = new File(path);if (!current.exists() || !current.isDirectory()) {return;}for (File child : current.listFiles()) {if (!child.isDirectory()) {continue;}collector.add(child.getAbsolutePath());collectDirs(child.getAbsolutePath(), collector);}}private static List<URL> doGetURLs(final String path) {Validate.isTrue(!StringUtils.isBlank(path), "jar包路径不能为空.");File jarPath = new File(path);Validate.isTrue(jarPath.exists() && jarPath.isDirectory(),"jar包路径必须存在且为目录.");/* set filter */FileFilter jarFilter = new FileFilter() {@Overridepublic boolean accept(File pathname) {return pathname.getName().endsWith(".jar");}};/* iterate all jar */File[] allJars = new File(path).listFiles(jarFilter);List<URL> jarURLs = new ArrayList<URL>(allJars.length);for (int i = 0; i < allJars.length; i++) {try {jarURLs.add(allJars[i].toURI().toURL());} catch (Exception e) {throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR,"系统加载jar包出错", e);}}return jarURLs;}
}

• LoadUtil
  LoadUtil 是一个工具类，用于辅助插件的加载和初始化过程。LoadUtil 类通常包含静态方法，这些方法简化了插件加载的逻辑，使得 DataX 的核心框架能够与具体的插件进行交互。
  LoadUtil 的主要职责包括：
  插件加载：LoadUtil 提供了加载插件的方法。这些方法会根据配置文件中指定的插件类型和名称，使用 Java 的反射机制来加载插件的类定义。加载过程可能包括查找类路径下的 JAR 文件、读取插件的元数据以及验证插件的合法性。
  插件实例化：一旦插件类被加载，LoadUtil 会负责创建插件的实例。这通常涉及到调用插件类的无参构造函数，并返回该实例的引用。LoadUtil 会处理任何与实例化相关的异常，以确保在出现问题时能够给出适当的错误消息。
  插件注册：加载并实例化插件后，LoadUtil 可能会将插件实例注册到一个全局的插件注册中心。这样，DataX 的其他部分就可以在需要时获取并使用这些插件实例。
  配置传递：LoadUtil 还可能负责将配置文件中针对插件的配置参数传递给插件实例。这确保了插件能够根据用户的配置进行正确的初始化。
  错误处理：如果在加载、实例化或配置插件过程中发生错误，LoadUtil 会负责处理这些错误。这可能包括记录日志、抛出异常或采取其他恢复措施。

public class LoadUtil {private static final String pluginTypeNameFormat = "plugin.%s.%s";private LoadUtil() {}private enum ContainerType {Job("Job"), Task("Task");private String type;private ContainerType(String type) {this.type = type;}public String value() {return type;}}/*** 所有插件配置放置在pluginRegisterCenter中，为区别reader、transformer和writer，还能区别* 具体pluginName，故使用pluginType.pluginName作为key放置在该map中*/private static Configuration pluginRegisterCenter;/*** jarLoader的缓冲*/private static Map<String, JarLoader> jarLoaderCenter = new HashMap();/*** 设置pluginConfigs，方便后面插件来获取** @param pluginConfigs*/public static void bind(Configuration pluginConfigs) {pluginRegisterCenter = pluginConfigs;}private static String generatePluginKey(PluginType pluginType,String pluginName) {return String.format(pluginTypeNameFormat, pluginType.toString(),pluginName);}private static Configuration getPluginConf(PluginType pluginType,String pluginName) {Configuration pluginConf = pluginRegisterCenter.getConfiguration(generatePluginKey(pluginType, pluginName));if (null == pluginConf) {throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INSTALL_ERROR,String.format("DataX不能找到插件[%s]的配置.",pluginName));}return pluginConf;}/*** 加载JobPlugin，reader、writer都可能要加载** @param pluginType* @param pluginName* @return*/public static AbstractJobPlugin loadJobPlugin(PluginType pluginType,String pluginName) {Class<? extends AbstractPlugin> clazz = LoadUtil.loadPluginClass(pluginType, pluginName, ContainerType.Job);try {AbstractJobPlugin jobPlugin = (AbstractJobPlugin) clazz.newInstance();jobPlugin.setPluginConf(getPluginConf(pluginType, pluginName));return jobPlugin;} catch (Exception e) {throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format("DataX找到plugin[%s]的Job配置.",pluginName), e);}}/*** 加载taskPlugin，reader、writer都可能加载** @param pluginType* @param pluginName* @return*/public static AbstractTaskPlugin loadTaskPlugin(PluginType pluginType,String pluginName) {Class<? extends AbstractPlugin> clazz = LoadUtil.loadPluginClass(pluginType, pluginName, ContainerType.Task);try {AbstractTaskPlugin taskPlugin = (AbstractTaskPlugin) clazz.newInstance();taskPlugin.setPluginConf(getPluginConf(pluginType, pluginName));return taskPlugin;} catch (Exception e) {throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format("DataX不能找plugin[%s]的Task配置.",pluginName), e);}}/*** 根据插件类型、名字和执行时taskGroupId加载对应运行器** @param pluginType* @param pluginName* @return*/public static AbstractRunner loadPluginRunner(PluginType pluginType, String pluginName) {AbstractTaskPlugin taskPlugin = LoadUtil.loadTaskPlugin(pluginType,pluginName);switch (pluginType) {case READER:return new ReaderRunner(taskPlugin);case WRITER:return new WriterRunner(taskPlugin);default:throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format("插件[%s]的类型必须是[reader]或[writer]!",pluginName));}}/*** 反射出具体plugin实例** @param pluginType* @param pluginName* @param pluginRunType* @return*/@SuppressWarnings("unchecked")private static synchronized Class<? extends AbstractPlugin> loadPluginClass(PluginType pluginType, String pluginName,ContainerType pluginRunType) {Configuration pluginConf = getPluginConf(pluginType, pluginName);JarLoader jarLoader = LoadUtil.getJarLoader(pluginType, pluginName);try {return (Class<? extends AbstractPlugin>) jarLoader.loadClass(pluginConf.getString("class") + "$"+ pluginRunType.value());} catch (Exception e) {throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR, e);}}public static synchronized JarLoader getJarLoader(PluginType pluginType,String pluginName) {Configuration pluginConf = getPluginConf(pluginType, pluginName);JarLoader jarLoader = jarLoaderCenter.get(generatePluginKey(pluginType,pluginName));if (null == jarLoader) {String pluginPath = pluginConf.getString("path");if (StringUtils.isBlank(pluginPath)) {throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,String.format("%s插件[%s]路径非法!",pluginType, pluginName));}jarLoader = new JarLoader(new String[]{pluginPath});jarLoaderCenter.put(generatePluginKey(pluginType, pluginName),jarLoader);}return jarLoader;}
}

• ClassLoaderSwapper

ClassLoaderSwapper有一个属性storeClassLoader， 用于保存着当前线程的classLoader切换之前的ClassLoader。

/*** 为避免jar冲突，比如hbase可能有多个版本的读写依赖jar包，JobContainer和TaskGroupContainer，就需要脱离当前classLoader去加载这些jar包，执行完成后，又退回到原来classLoader上继续执行接下来的代码*/
public final class ClassLoaderSwapper {private ClassLoader storeClassLoader = null;private ClassLoaderSwapper() {}public static ClassLoaderSwapper newCurrentThreadClassLoaderSwapper() {return new ClassLoaderSwapper();}/*** 保存当前classLoader，并将当前线程的classLoader设置为所给classLoader** @param* @return*/public ClassLoader setCurrentThreadClassLoader(ClassLoader classLoader) {this.storeClassLoader = Thread.currentThread().getContextClassLoader();Thread.currentThread().setContextClassLoader(classLoader);return this.storeClassLoader;}/*** 将当前线程的类加载器设置为保存的类加载* @return*/public ClassLoader restoreCurrentThreadClassLoader() {ClassLoader classLoader = Thread.currentThread().getContextClassLoader();Thread.currentThread().setContextClassLoader(this.storeClassLoader);return classLoader;}
}