夜雨聆风系列：DataX 源码解析 04
1. 为什么写这篇
前面三篇，我们把 DataX 的主流程、调度流程、数据传输流程都拆了一遍。
现在还剩最后一个问题：
DataX 为什么可以支持这么多 Reader 和 Writer？
它可以读 MySQL、Oracle、HDFS、Hive，也可以写 Doris、HBase、FTP 等各种目标端。对于使用者来说，只是在 JSON 里改一个 reader/writer 名称。
但从框架角度看，这背后一定要解决几个问题：
根据配置找到对应插件。
加载插件自己的 jar 包。
创建 Reader.Job / Writer.Job 实例。
避免不同插件之间依赖冲突。
初始化插件参数，然后交给后续 split、schedule、task 执行。
所以这篇作为 DataX 系列收尾，专门看插件是怎么加载起来的。
2. 我遇到的问题
如果我们写过插件化系统，就会知道“能扩展”不是一句口号。
真正难的是：
插件放在哪里？
插件类名从哪里来？
插件依赖怎么加载？
如果两个插件依赖不同版本 jar，怎么隔离？
加载完之后，框架如何统一调用？
DataX 没有依赖很重的框架，而是用一套 Java 原生机制完成插件加载。
这一套机制的核心，是 LoadUtil 和自定义类加载器 JarLoader。
我读这部分源码时，关注的是这条线：
JobContainer.init() -> initJobReader / initJobWriter -> LoadUtil.getJarLoader() -> LoadUtil.loadJobPlugin() -> LoadUtil.loadPluginClass() -> JarLoader.loadClass()理解这条线，就能明白 DataX 的插件体系为什么成立。
3. 我的解决思路
我把插件加载拆成 5 步：
从任务配置里拿到 reader/writer 插件名称。
根据插件类型和名称找到插件配置。
根据插件路径创建或复用 JarLoader。
通过 JarLoader 加载插件类。
实例化插件，并调用插件自己的 init 方法。
这里最关键的不是“反射 new 一个对象”，而是 类加载器隔离。
如果所有插件都丢给主程序 ClassLoader 加载，不同插件之间的依赖就很容易互相污染。
DataX 的做法是：每个插件可以有自己的 JarLoader。加载 Reader 插件时，把当前线程上下文类加载器切到对应 Reader 的 JarLoader；加载完成后，再恢复原来的类加载器。
这就让框架主程序和插件代码之间形成了比较清晰的边界。
4. 核心实现
插件加载从 JobContainer.start() 里的 init() 开始。
在 init() 中，DataX 会先初始化 Reader，再初始化 Writer：
this.jobReader = this.initJobReader(jobPluginCollector);this.jobWriter = this.initJobWriter(jobPluginCollector);进入 initJobReader 后，第一步是从配置中拿到 reader 插件名称：
this.readerPluginName = this.configuration.getString( CoreConstant.DATAX_JOB_CONTENT_READER_NAME);然后根据插件类型和插件名拿到对应的 JarLoader，并切换当前线程的 ClassLoader：
classLoaderSwapper.setCurrentThreadClassLoader( LoadUtil.getJarLoader(PluginType.READER, this.readerPluginName));这一步非常关键。
它意味着接下来加载 Reader 插件类时，会进入 Reader 插件自己的类加载环境。
然后 DataX 调用：
Reader.Job jobReader = (Reader.Job) LoadUtil.loadJobPlugin( PluginType.READER, this.readerPluginName);loadJobPlugin 里面会继续调用 loadPluginClass：
Class<? extends AbstractPlugin> clazz = LoadUtil.loadPluginClass(pluginType, pluginName, ContainerType.Job);真正加载类的地方在这里：
return (Class<? extends AbstractPlugin>) jarLoader.loadClass( pluginConf.getString("class") + "$" + pluginRunType.value());这个类名拼接也很有意思。
插件配置里的 class 指向插件主类，比如某个 Reader。DataX 再根据运行阶段拼上 $Job 或 $Task，加载对应的内部类。
也就是说，插件不是只有一个实现，而是按运行阶段分成 Job 和 Task 两层。
Job 阶段负责初始化、切分等逻辑。
Task 阶段负责真正执行读写。
接下来再看 getJarLoader。
DataX 会根据插件类型和插件名生成 key，从缓存里找 JarLoader：
JarLoader jarLoader = jarLoaderCenter.get( generatePluginKey(pluginType, pluginName));如果缓存里没有，就根据插件配置里的 path 创建一个新的 JarLoader：
String pluginPath = pluginConf.getString("path");jarLoader = new JarLoader(new String[]{pluginPath});jarLoaderCenter.put(generatePluginKey(pluginType, pluginName), jarLoader);这里做了两件事：
同一个插件的 JarLoader 会被缓存，避免重复创建。
每个插件可以按自己的路径加载 jar，实现一定程度的依赖隔离。
最后看 JarLoader。
它继承自 URLClassLoader：
public class JarLoader extends URLClassLoader创建时会扫描插件路径、子路径，以及目录下的 jar 文件，把它们加入 classpath。
这样插件目录里的 jar 包就能被这个 JarLoader 加载。
整个插件加载流程可以总结成：
配置里写插件名 -> 找到插件配置和路径 -> 为插件创建 JarLoader -> 切换线程上下文 ClassLoader -> 反射加载插件 Job/Task 类 -> 初始化插件参数 -> 恢复原 ClassLoader这就是 DataX 插件机制的核心。
5. 踩坑记录
这部分源码里，最容易误解的点有三个。
第一个误解：以为插件加载只是反射。
反射只是最后一步。真正重要的是在反射之前，DataX 已经为插件准备好了对应的 JarLoader，并切换了线程上下文类加载器。
第二个误解：以为 Reader 和 Writer 只有一个插件类。
DataX 插件通常分 Job 和 Task 两层。Job 负责全局阶段，比如初始化、切分；Task 负责具体执行，比如每个分片的读写。
第三个误解：忽略 JarLoader 缓存。
如果每次加载插件都重新创建类加载器，成本会变高，也容易出现类身份不一致的问题。DataX 用 jarLoaderCenter 缓存同一个插件的 JarLoader，避免重复创建。
还有一个细节也值得注意：DataX 在加载插件前切换 ClassLoader，加载完后会恢复原来的 ClassLoader。
这一步如果漏掉，后续主程序或其他插件加载类时就可能进入错误的类加载环境。
6. 总结
DataX 插件加载机制可以提炼成 5 条：
插件名称来自任务配置中的 reader/writer。
插件类名和路径来自插件配置。
JarLoader 负责把插件目录和 jar 加入 classpath。
LoadUtil 负责查配置、拿 JarLoader、加载插件类并实例化。
ClassLoader 切换和恢复保证插件加载过程不污染主流程。
这套机制并不花哨，但很实用。
DataX 没有把 Reader 和 Writer 写死在框架里，而是通过配置、反射、类加载器隔离，把不同数据源接入框架。
到这里，DataX 系列四篇就能连起来了：
整体架构：Job / Task / TaskGroup / Channel调度流程：Task 如何分配并启动数据传输：Reader 和 Writer 如何交换 Record插件加载：Reader 和 Writer 如何被加载进框架这四条线合在一起，就是 DataX 作为离线同步框架最核心的设计。
7. 延伸阅读
个人网站原文：https://yuting0907.github.io/posts/2025/09/cf5c40b3.html
第一篇：DataX 是怎么把 100 张表同步起来的？
第二篇：DataX 一个任务到底是怎么被调度起来的？
第三篇：DataX 的 Reader 和 Writer 是怎么把数据传起来的？
GitHub 源码仓库：https://github.com/YUTING0907/YUTING0907.github.io