github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-02 15:12:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

不依赖于user.dir环境变量,所以插件加载工厂便没有意义了

Browse Source 
不依赖于user.dir环境变量,所以插件加载工厂便没有意义了
This commit is contained in:
FuYouJ 2023-08-05 23:11:43 +08:00
parent 03cfbfa2f7
commit af631a0a00
11 changed files with 271 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
core/src/main/java/com/alibaba/datax/core/util/container/LoadUtil.java
View File

@ -49,7 +49,7 @@ public class LoadUtil {
/**
* jarLoader的缓冲
*/
private static Map<String, ClassLoader> jarLoaderCenter = new HashMap();
private static Map<String, JarLoader> jarLoaderCenter = new HashMap();
/**
* 设置pluginConfigs，方便后面插件来获取
@ -167,7 +167,7 @@ public class LoadUtil {
PluginType pluginType, String pluginName,
ContainerType pluginRunType) {
Configuration pluginConf = getPluginConf(pluginType, pluginName);
ClassLoader jarLoader = LoadUtil.getJarLoader(pluginType, pluginName);
JarLoader jarLoader = LoadUtil.getJarLoader(pluginType, pluginName);
try {
return (Class<? extends AbstractPlugin>) jarLoader
.loadClass(pluginConf.getString("class") + "$"
@ -177,14 +177,22 @@ public class LoadUtil {
}
}
public static synchronized ClassLoader getJarLoader(PluginType pluginType,
public static synchronized JarLoader getJarLoader(PluginType pluginType,
String pluginName) {
Configuration pluginConf = getPluginConf(pluginType, pluginName);
ClassLoader jarLoader = jarLoaderCenter.get(generatePluginKey(pluginType,
JarLoader jarLoader = jarLoaderCenter.get(generatePluginKey(pluginType,
pluginName));
if (null == jarLoader) {
jarLoader = PluginLoaderFactory.create(pluginConf,pluginType,pluginName);
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);
}

76
datax-example/doc/README.md
View File

@ -21,28 +21,82 @@
### 实现原理
- 不修改原有的ConfigParer,使用新的ExampleConfigParser,仅用于example模块。
- 提供新的PluginLoader 插件加载器，可以从程序运行目录获取插件，与JarLoader各司其职。
- 不修改原有的ConfigParer,使用新的ExampleConfigParser,仅用于example模块。他不依赖datax.home,而是依赖ide编译后的target目录
- 将ide的target目录作为每个插件的目录类加载目录。
![img](img/img02.png)
### 如何使用
1.修改插件的pom文件，做如下改动。以streamreader为例。<br/>
改动前
```xml
<build>
<plugins>
<!-- compiler plugin -->
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>${jdk-version}</source>
<target>${jdk-version}</target>
<encoding>${project-sourceEncoding}</encoding>
</configuration>
</plugin>
</plugins>
</build>
```
改动后
```xml
<build>
<resources>
<!--将resource目录也输出到target-->
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.*</include>
</includes>
<filtering>true</filtering>
</resource>
</resources>
<plugins>
<!-- compiler plugin -->
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>${jdk-version}</source>
<target>${jdk-version}</target>
<encoding>${project-sourceEncoding}</encoding>
</configuration>
</plugin>
</plugins>
</build>
```
2.在datax-example模块引入你需要的插件，默认只引入了streamreader、writer
3.打开datax-example的Main class
```java
public class Main {
public static void main(String[] args) {
//1. 在 datax-example pom文件中加入测试插件模块的依赖，默认导入了streamreader/writer
//2. 在此处指定你的测试文件路径
String path = "/job/stream2stream.json";
Configuration configuration = ExampleConfigParser.parse(
PathUtil.getAbsolutePathFromClassPath(path)
);
/**
* 注意！
* 1.在example模块pom文件添加你依赖的的调试插件，
* 你可以直接打开本模块的pom文件,参考是如何引入streamreader，streamwriter
* 2. 在此处指定你的job文件
*/
public static void main(String[] args) {
String classPathJobPath = "/job/stream2stream.json";
String absJobPath = PathUtil.getAbsolutePathFromClassPath(classPathJobPath);
startExample(absJobPath);
}
public static void startExample(String jobPath) {
Configuration configuration = ExampleConfigParser.parse(jobPath);
Engine engine = new Engine();
engine.start(configuration);
}
}
```
### 注意
此模块不参与打包

BIN
datax-example/doc/img/img02.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 61 KiB

After

Width:  |  Height:  |  Size: 66 KiB

24
datax-example/pom.xml
View File

@ -28,6 +28,7 @@
<artifactId>datax-core</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<!-- 默认引入streamwriter,streamreader-->
<dependency>
<groupId>com.alibaba.datax</groupId>
<artifactId>streamwriter</artifactId>
@ -49,12 +50,31 @@
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<directory>src/main/resources</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.*</include>
</includes>
<filtering>true</filtering>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.*</include>
</includes>
<filtering>true</filtering>
</resource>
</resources>
<plugins>
<!-- compiler plugin -->
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>${jdk-version}</source>
<target>${jdk-version}</target>
<encoding>${project-sourceEncoding}</encoding>
</configuration>
</plugin>
</plugins>
</build>
</project>

22
datax-example/src/main/java/com/alibaba/datax/example/Main.java
View File

@ -10,15 +10,25 @@ import com.alibaba.datax.example.util.PathUtil;
* @author fuyouj
*/
public class Main {
public static void main(String[] args) {
//在此处指定你的测试文件路径
String path = "/job/stream2stream.json";
Configuration configuration = ExampleConfigParser.parse(
PathUtil.getAbsolutePathFromClassPath(path)
);
/**
* 1.在example模块pom文件添加你依赖的的调试插件，
* 你可以直接打开本模块的pom文件,参考是如何引入streamreader，streamwriter
* 2. 在此处指定你的job文件
*/
public static void main(String[] args) {
String classPathJobPath = "/job/stream2stream.json";
String absJobPath = PathUtil.getAbsolutePathFromClassPath(classPathJobPath);
startExample(absJobPath);
}
public static void startExample(String jobPath) {
Configuration configuration = ExampleConfigParser.parse(jobPath);
Engine engine = new Engine();
engine.start(configuration);
}
}

71
datax-example/src/main/java/com/alibaba/datax/example/util/ExampleConfigParser.java
View File

@ -7,10 +7,10 @@ import com.alibaba.datax.core.util.FrameworkErrorCode;
import com.alibaba.datax.core.util.container.CoreConstant;
import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;
import java.util.*;
/**
* @author fuyouj
@ -44,9 +44,15 @@ public class ExampleConfigParser {
Configuration configuration = Configuration.newDefault();
String workingDirectory = System.getProperty("user.dir");
File file = new File(workingDirectory);
scanPlugin(configuration, file.listFiles(), pluginTypeMap);
//最初打算通过user.dir获取工作目录来扫描插件，
//但是user.dir在不同有一些不确定性，所以废弃了这个选择
for (File basePackage : runtimeBasePackages()) {
if (pluginTypeMap.isEmpty()) {
break;
}
scanPluginByPackage(basePackage, configuration, basePackage.listFiles(), pluginTypeMap);
}
if (!pluginTypeMap.isEmpty()) {
throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR,
"load plugin failed，未完成指定插件加载:"
@ -55,7 +61,42 @@ public class ExampleConfigParser {
return configuration;
}
private static void scanPlugin(Configuration configuration, File[] files, Map<String, String> pluginTypeMap) {
/**
* 通过classLoader获取程序编译的输出目录
* @return File[/datax-example/target/classes,xxReader/target/classes,xxWriter/target/classes]
*/
private static File[] runtimeBasePackages() {
List<File> basePackages = new ArrayList<>();
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Enumeration<URL> resources = null;
try {
resources = classLoader.getResources("");
} catch (IOException e) {
throw DataXException.asDataXException(e.getMessage());
}
while (resources.hasMoreElements()) {
URL resource = resources.nextElement();
File file = new File(resource.getFile());
if (file.isDirectory()) {
basePackages.add(file);
}
}
return basePackages.toArray(new File[0]);
}
/**
*
* @param packageFile 编译出来的target/classes根目录 便于找到插件时设置插件的URL目录，设置根目录是最保险的方式
* @param configuration pluginConfig
* @param files 待扫描文件
* @param needPluginTypeMap 需要的插件
*/
private static void scanPluginByPackage(File packageFile,
Configuration configuration,
File[] files,
Map<String, String> needPluginTypeMap) {
if (files == null) {
return;
}
@ -64,22 +105,26 @@ public class ExampleConfigParser {
Configuration pluginDesc = Configuration.from(file);
String descPluginName = pluginDesc.getString("name", "");
if (pluginTypeMap.containsKey(descPluginName)) {
if (needPluginTypeMap.containsKey(descPluginName)) {
String type = pluginTypeMap.get(descPluginName);
configuration.merge(parseOnePlugin(type, descPluginName, pluginDesc), false);
pluginTypeMap.remove(descPluginName);
String type = needPluginTypeMap.get(descPluginName);
configuration.merge(parseOnePlugin(packageFile.getAbsolutePath(),type, descPluginName, pluginDesc), false);
needPluginTypeMap.remove(descPluginName);
}
} else {
scanPlugin(configuration, file.listFiles(), pluginTypeMap);
scanPluginByPackage(packageFile,configuration, file.listFiles(), needPluginTypeMap);
}
}
}
private static Configuration parseOnePlugin(String pluginType, String pluginName, Configuration pluginDesc) {
pluginDesc.set("loadType","pluginLoader");
private static Configuration parseOnePlugin(String packagePath,
String pluginType,
String pluginName,
Configuration pluginDesc) {
//设置path 兼容jarLoader的加载方式URLClassLoader
pluginDesc.set("path", packagePath);
Configuration pluginConfInJob = Configuration.newDefault();
pluginConfInJob.set(
String.format("plugin.%s.%s", pluginType, pluginName),

14
datax-example/src/test/java/com/alibaba/datax/example/DataXExampleTest.java
View File

@ -1,24 +1,16 @@
package com.alibaba.datax.example;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.Engine;
import com.alibaba.datax.example.util.ExampleConfigParser;
import com.alibaba.datax.example.util.PathUtil;
import org.junit.Test;
public class DataXExampleTest extends ExampleTestTemplate {
public class DataXExampleTest {
@Test
public void testStreamReader2StreamWriter() {
String path = "/job/stream2stream.json";
Configuration testConfiguration = ExampleConfigParser.parse(
PathUtil.getAbsolutePathFromClassPath(path)
);
Engine engine = new Engine();
engine.start(testConfiguration);
String jobPath = PathUtil.getAbsolutePathFromClassPath(path);
Main.startExample(jobPath);
}
}

3
datax-example/src/test/java/com/alibaba/datax/example/util/com/alibaba/datax/example/util/ExampleConfigParserTest.java
View File

@ -1,7 +1,6 @@
package com.alibaba.datax.example.util.com.alibaba.datax.example.util;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.example.ExampleTestTemplate;
import com.alibaba.datax.example.util.ExampleConfigParser;
import com.alibaba.datax.example.util.PathUtil;
import org.junit.Assert;
@ -10,7 +9,7 @@ import org.junit.Test;
import java.io.File;
public class ExampleConfigParserTest extends ExampleTestTemplate {
public class ExampleConfigParserTest {
@Test

142
pom.xml
View File

@ -47,88 +47,88 @@
<module>transformer</module>
<!-- reader -->
<module>mysqlreader</module>
<module>drdsreader</module>
<module>sqlserverreader</module>
<module>postgresqlreader</module>
<module>kingbaseesreader</module>
<module>oraclereader</module>
<module>cassandrareader</module>
<module>oceanbasev10reader</module>
<module>rdbmsreader</module>
<!-- <module>mysqlreader</module>-->
<!-- <module>drdsreader</module>-->
<!-- <module>sqlserverreader</module>-->
<!-- <module>postgresqlreader</module>-->
<!-- <module>kingbaseesreader</module>-->
<!-- <module>oraclereader</module>-->
<!-- <module>cassandrareader</module>-->
<!-- <module>oceanbasev10reader</module>-->
<!-- <module>rdbmsreader</module>-->
<module>odpsreader</module>
<module>otsreader</module>
<module>otsstreamreader</module>
<module>hbase11xreader</module>
<module>hbase094xreader</module>
<module>hbase11xsqlreader</module>
<module>hbase20xsqlreader</module>
<!-- <module>odpsreader</module>-->
<!-- <module>otsreader</module>-->
<!-- <module>otsstreamreader</module>-->
<!-- <module>hbase11xreader</module>-->
<!-- <module>hbase094xreader</module>-->
<!-- <module>hbase11xsqlreader</module>-->
<!-- <module>hbase20xsqlreader</module>-->
<module>ossreader</module>
<module>hdfsreader</module>
<module>ftpreader</module>
<module>txtfilereader</module>
<!-- <module>ossreader</module>-->
<!-- <module>hdfsreader</module>-->
<!-- <module>ftpreader</module>-->
<!-- <module>txtfilereader</module>-->
<module>streamreader</module>
<module>clickhousereader</module>
<!-- <module>clickhousereader</module>-->
<module>mongodbreader</module>
<module>tdenginereader</module>
<module>gdbreader</module>
<module>tsdbreader</module>
<module>opentsdbreader</module>
<module>loghubreader</module>
<module>datahubreader</module>
<module>starrocksreader</module>
<!-- <module>mongodbreader</module>-->
<!-- <module>tdenginereader</module>-->
<!-- <module>gdbreader</module>-->
<!-- <module>tsdbreader</module>-->
<!-- <module>opentsdbreader</module>-->
<!-- <module>loghubreader</module>-->
<!-- <module>datahubreader</module>-->
<!-- <module>starrocksreader</module>-->
<!-- writer -->
<module>mysqlwriter</module>
<module>starrockswriter</module>
<module>drdswriter</module>
<module>databendwriter</module>
<module>oraclewriter</module>
<module>sqlserverwriter</module>
<module>postgresqlwriter</module>
<module>kingbaseeswriter</module>
<module>adswriter</module>
<module>oceanbasev10writer</module>
<module>adbpgwriter</module>
<module>hologresjdbcwriter</module>
<module>rdbmswriter</module>
<!-- &lt;!&ndash; writer &ndash;&gt;-->
<!-- <module>mysqlwriter</module>-->
<!-- <module>starrockswriter</module>-->
<!-- <module>drdswriter</module>-->
<!-- <module>databendwriter</module>-->
<!-- <module>oraclewriter</module>-->
<!-- <module>sqlserverwriter</module>-->
<!-- <module>postgresqlwriter</module>-->
<!-- <module>kingbaseeswriter</module>-->
<!-- <module>adswriter</module>-->
<!-- <module>oceanbasev10writer</module>-->
<!-- <module>adbpgwriter</module>-->
<!-- <module>hologresjdbcwriter</module>-->
<!-- <module>rdbmswriter</module>-->
<module>odpswriter</module>
<module>osswriter</module>
<module>otswriter</module>
<module>hbase11xwriter</module>
<module>hbase094xwriter</module>
<module>hbase11xsqlwriter</module>
<module>hbase20xsqlwriter</module>
<module>kuduwriter</module>
<module>ftpwriter</module>
<module>hdfswriter</module>
<module>txtfilewriter</module>
<!-- <module>odpswriter</module>-->
<!-- <module>osswriter</module>-->
<!-- <module>otswriter</module>-->
<!-- <module>hbase11xwriter</module>-->
<!-- <module>hbase094xwriter</module>-->
<!-- <module>hbase11xsqlwriter</module>-->
<!-- <module>hbase20xsqlwriter</module>-->
<!-- <module>kuduwriter</module>-->
<!-- <module>ftpwriter</module>-->
<!-- <module>hdfswriter</module>-->
<!-- <module>txtfilewriter</module>-->
<module>streamwriter</module>
<module>elasticsearchwriter</module>
<module>mongodbwriter</module>
<module>tdenginewriter</module>
<module>ocswriter</module>
<module>tsdbwriter</module>
<module>gdbwriter</module>
<module>oscarwriter</module>
<module>loghubwriter</module>
<module>datahubwriter</module>
<module>cassandrawriter</module>
<module>clickhousewriter</module>
<module>doriswriter</module>
<module>selectdbwriter</module>
<module>adbmysqlwriter</module>
<module>neo4jwriter</module>
<!-- <module>elasticsearchwriter</module>-->
<!-- <module>mongodbwriter</module>-->
<!-- <module>tdenginewriter</module>-->
<!-- <module>ocswriter</module>-->
<!-- <module>tsdbwriter</module>-->
<!-- <module>gdbwriter</module>-->
<!-- <module>oscarwriter</module>-->
<!-- <module>loghubwriter</module>-->
<!-- <module>datahubwriter</module>-->
<!-- <module>cassandrawriter</module>-->
<!-- <module>clickhousewriter</module>-->
<!-- <module>doriswriter</module>-->
<!-- <module>selectdbwriter</module>-->
<!-- <module>adbmysqlwriter</module>-->
<!-- <module>neo4jwriter</module>-->
<!-- common support module -->
<module>plugin-rdbms-util</module>
<module>plugin-unstructured-storage-util</module>
<!-- <module>plugin-rdbms-util</module>-->
<!-- <module>plugin-unstructured-storage-util</module>-->
<module>datax-example</module>
</modules>

10
streamreader/pom.xml
View File

@ -39,6 +39,16 @@
</dependencies>
<build>
<resources>
<!--将resource目录也输出到target-->
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.*</include>
</includes>
<filtering>true</filtering>
</resource>
</resources>
<plugins>
<!-- compiler plugin -->
<plugin>

10
streamwriter/pom.xml
View File

@ -34,6 +34,16 @@
</dependencies>
<build>
<resources>
<!-- 将resource目录也输出到target-->
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.*</include>
</includes>
<filtering>true</filtering>
</resource>
</resources>
<plugins>
<!-- compiler plugin -->
<plugin>