datax(25):插件加载原理

大家好，又见面了，我是你们的朋友全栈君。

一、插件分类

按照功能分

1. reader， 读插件，例如mysqlReader，从mysql读取数据
2. writer， 写插件。例如mysqlWriter，给mysql写入数据；
3. transformer， 中间结果转换，例如SubstrTransformer用于字符截取；
4. handler， 主要用于任务执行前的准备工作和完成的收尾工作。

插件类型由PluginType枚举表示

public enum PluginType { 
   
	READER("reader"), TRANSFORMER("transformer"), WRITER("writer"), HANDLER("handler");
}

按照运行类型

1. Job级别的插件
2. Task级别的插件

二、插件目录结构

大目录：xxx\DataX\target\datax\datax\plugin下分2个reader和writer目录，下面以mysql为例

如上图，可以看到

• libs文件夹主要放该插件运行所依赖的jars
• xxx-xxx.jar即本插件最后打成的包
• plugin.json里面是本插件的元数据信息（名称，主类，描述信息，开发者），具体如下

{ 
   
    "name": "mysqlreader",
    "class": "com.alibaba.datax.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": "alibaba"
}

• plugin_job_template.json是本插件的一个模板JSON，加载时候会根据该模板里的参数校验用户的入参json

三、插件加载原理

1. 配置job信息，获取所有插件名称

给ConfigParser.parse(final String jobPath)传入job路径，该方法组装解析，最后返回一个conf对象，conf里解析出了reader，writer，handler等插件名称；

public static Configuration parse(final String jobPath) { 

Configuration configuration = parseJobConfig(jobPath);
configuration.merge(parseCoreConfig(CoreConstant.DATAX_CONF_PATH), false);
// todo config优化，只捕获需要的plugin
String readerPluginName = configuration.getString(CoreConstant.DATAX_JOB_CONTENT_READER_NAME);
String writerPluginName = configuration.getString(CoreConstant.DATAX_JOB_CONTENT_WRITER_NAME);
String preHandlerName = configuration.getString(CoreConstant.DATAX_JOB_PREHANDLER_PLUGINNAME);
String postHandlerName = configuration.getString(CoreConstant.DATAX_JOB_POSTHANDLER_PLUGINNAME);
Set<String> pluginList = new HashSet<>();
pluginList.add(readerPluginName);
pluginList.add(writerPluginName);
if (StringUtils.isNotEmpty(preHandlerName)) { 

pluginList.add(preHandlerName);
}
if (StringUtils.isNotEmpty(postHandlerName)) { 

pluginList.add(postHandlerName);
}
try { 

configuration.merge(parsePluginConfig(new ArrayList<>(pluginList)), false);
} catch (Exception e) { 

//吞掉异常，保持log干净。这里message足够。
LOG.warn(String
.format("插件[%s,%s]加载失败，1s后重试... Exception:%s ", readerPluginName, writerPluginName,
e.getMessage()));
try { 

Thread.sleep(1000);
} catch (InterruptedException e1) { 

//
}
configuration.merge(parsePluginConfig(new ArrayList<>(pluginList)), false);
}
return configuration;
}

2. 根据插件名称获取插件配置（plugin.json）

上一步获取了很多插件信息，本步骤根据插件名称和datax的规范，从目录中获取每个插件的详细信息；

public static Configuration parsePluginConfig(List<String> wantPluginNames) { 

Configuration configuration = Configuration.newDefault();
Set<String> replicaCheckPluginSet = new HashSet<>();
int complete = 0;
for (final String each : ConfigParser
.getDirAsList(CoreConstant.DATAX_PLUGIN_READER_HOME)) { 

Configuration eachReaderConfig = ConfigParser
.parseOnePluginConfig(each, "reader", replicaCheckPluginSet, wantPluginNames);
if (eachReaderConfig != null) { 

configuration.merge(eachReaderConfig, true);
complete += 1;
}
}
for (final String each : ConfigParser
.getDirAsList(CoreConstant.DATAX_PLUGIN_WRITER_HOME)) { 

Configuration eachWriterConfig = ConfigParser
.parseOnePluginConfig(each, "writer", replicaCheckPluginSet, wantPluginNames);
if (eachWriterConfig != null) { 

configuration.merge(eachWriterConfig, true);
complete += 1;
}
}
if (wantPluginNames != null && wantPluginNames.size() > 0
&& wantPluginNames.size() != complete) { 

throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR,
"插件加载失败，未完成指定插件加载:" + wantPluginNames);
}
return configuration;
}

3.动态加载插件

插件的加载都是使用ClassLoader动态加载。 为了避免类的冲突，datax通过自定义类加载器JarLoader，提供Jar隔离的加载机制。

• 加载器由JarLoader实现
• 插件的加载接口由LoadUtil类负责
• 当要加载一个插件时，需要实例化一个JarLoader，然后切换thread class loader之后，才加载插件。这个主要由ClassLoaderSwapper实现

3.1 JarLoader 类

JarLoader继承URLClassLoader，扩充了可以加载目录的功能。可以从指定的目录下，把传入的路径、及其子路径、以及路径中的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 = pathname -> pathname.getName().endsWith(".jar");
/* iterate all jar */
File[] allJars = new File(path).listFiles(jarFilter);
List<URL> jarURLs = new ArrayList<>(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;
}
}

3.2 LoadUtil 类

LoadUtil管理着插件的加载器，调用getJarLoader返回插件对应的加载器。

/** * Created by jingxing on 14-8-24. * <p/> * 插件加载器，大体上分reader、transformer（还未实现）和writer三种插件类型， reader和writer在执行时又可能出现Job和Task两种运行时（加载的类不同） */
public class LoadUtil { 

private static final String pluginTypeNameFormat = "plugin.%s.%s";
private LoadUtil() { 

}
private enum ContainerType { 

Job("Job"),
Task("Task");
private String type;
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，方便后面插件来获取 初始化PluginLoader，可以获取各种插件配置 * * @param pluginConfigs */
public static void bind(Configuration pluginConfigs) { 

pluginRegisterCenter = pluginConfigs;
}
/** * 根据插件类型+插件名称，生成一个 字符串。插件中心根据该字符串找到对应插件 * * @param pluginType PluginType * @param pluginName String * @return String */
private static String generatePluginKey(PluginType pluginType, String pluginName) { 

return String.format(pluginTypeNameFormat, pluginType.toString(), pluginName);
}
/** * 根据插件类型和插件名称，获取配置； <br/> 1 根据 插件类型+插件名称，返回string ； <br/> 2 从 pluginRegisterCenter 中根据 * string获取配置 * * @param pluginType * @param pluginName * @return */
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 type PluginType * @param name String * @return AbstractJobPlugin */
public static AbstractJobPlugin loadJobPlugin(PluginType type, String name) { 

Class<? extends AbstractPlugin> clazz = loadPluginClass(type, name, ContainerType.Job);
try { 

AbstractJobPlugin jobPlugin = (AbstractJobPlugin) clazz.newInstance();
jobPlugin.setPluginConf(getPluginConf(type, name));
return jobPlugin;
} catch (Exception e) { 

throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,
String.format("DataX找到plugin[%s]的Job配置.", name), e);
}
}
/** * 原理类同上面loadJobPlugin 方法。加载taskPlugin，reader、writer都可能加载 * * @param type PluginType * @param name String * @return AbstractTaskPlugin */
public static AbstractTaskPlugin loadTaskPlugin(PluginType type, String name) { 

Class<? extends AbstractPlugin> clz = LoadUtil.loadPluginClass(type, name, ContainerType.Task);
try { 

AbstractTaskPlugin taskPlugin = (AbstractTaskPlugin) clz.newInstance();
taskPlugin.setPluginConf(getPluginConf(type, name));
return taskPlugin;
} catch (Exception e) { 

throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,
String.format("DataX不能找plugin[%s]的Task配置.", name), 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;
}
}

3.3 ClassLoaderSwapper

切换类加载器

/** * Created by jingxing on 14-8-29. * * 为避免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（实现classLoader互换） * * @param * @return */
public ClassLoader setCurrentThreadClassLoader(ClassLoader classLoader) { 

// 1 将当前线程的 classLoader 先保持到 本类的 storeClassLoader 中
this.storeClassLoader = Thread.currentThread().getContextClassLoader();
// 2 将当前线程的 classLoader 设置为 传入的 classLoader
Thread.currentThread().setContextClassLoader(classLoader);
return this.storeClassLoader;
}
/** * 将当前线程的类加载器设置为保存的类加载 * @return */
public ClassLoader restoreCurrentThreadClassLoader() { 

ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(this.storeClassLoader);
return classLoader;
}
}