需求
内网的报告信息，需要在App上查看
内网环境：
- 一台
Windows Server，此服务器即是文件服务器，也是数据库服务器- 报告信息存储在D盘
- 数据库为
PostgreSQL
App环境：
- 接口服务在
阿里云ECS - 报告信息存储在
阿里云OSS，数据库为阿里云RDS MySQL
内网出报告后，将报告同步到阿里云OSS，并将PostgreSQL数据同步到MySQL
实现
- 数据库同步使用Kettle
- 文件同步使用commons-io的FileMonitor
数据库同步——Kettle
- 官网介绍
- 一个
开源的基于Java的企业级ETL工具，被Pentaho公司收购后，改名为PDIPentaho Data Integration
组成
术语
- Transformation 转换，是一系列的数据流，由
Step步骤和Hop跳组成，文件扩展名ktr - Step 步骤，具体的输入输出
- Hop 跳，是一个单向的数据通道，控制数据的流向
- Job 作业，任务调度+转换，文件扩展名
kjb
安装
- 北京大学镜像
- 下载
pdi-ce-8.2.0.0-342.zip - 解压后启动客户端UI
- Windows，双击Spoon.bat
- MacOS，双击spoon.sh
安装数据库驱动
PostgreSQL驱动
MySQL驱动
下载数据库驱动放置在/xxx/data-integration/lib下
编写转换
输入
- 新建>转换>输入>表输入
-
新建连接
-
编写SQL
输出
- 新建>转换>输出>插入 / 更新
-
新建连接
-
编写表间映射关系
表字段目标表的字段
流字段源数据表的字段
用来查询的关键字通过此条件匹配数据，匹配成功则更新数据，匹配失败则插入数据
更新字段设置要同步的字段，更新，Y表示该字段执行插入和更新，N表示该字段只执行插入
运行
编写作业
定时调度
- 新建>作业>通用>Start
-
30秒同步一次
转换
-
选择刚刚创建的转换
连线
- Shift选中
Start，拖拽到转换
运行
-
停止作业
文件同步——commons-io
原理
- FileAlterationMonitor开启线程轮询检查，默认轮询间隔10s
public void run() {
while (running) {
for (final FileAlterationObserver observer : observers) {
observer.checkAndNotify();
}
if (!running) {
break;
}
try {
Thread.sleep(interval);
} catch (final InterruptedException ignored) {
// ignore
}
}
}
public FileAlterationMonitor() {
this(10000);
}
- 根据文件名对比缓存的目录文件
previous和当前的目录文件files
private void checkAndNotify(final FileEntry parent, final FileEntry[] previous, final File[] files) {
int c = 0;
final FileEntry[] current = files.length > 0 ? new FileEntry[files.length] : FileEntry.EMPTY_ENTRIES;
for (final FileEntry entry : previous) {
while (c < files.length && comparator.compare(entry.getFile(), files[c]) > 0) {
current[c] = createFileEntry(parent, files[c]);
// 触发onDirectoryCreate/onFileCreate
doCreate(current[c]);
c++;
}
if (c < files.length && comparator.compare(entry.getFile(), files[c]) == 0) {
// 触发onDirectoryChange/onFileChange
doMatch(entry, files[c]);
checkAndNotify(entry, entry.getChildren(), listFiles(files[c]));
current[c] = entry;
c++;
} else {
checkAndNotify(entry, entry.getChildren(), FileUtils.EMPTY_FILE_ARRAY);
// 触发 onDirectoryDelete/onFileDelete
doDelete(entry);
}
}
for (; c < files.length; c++) {
current[c] = createFileEntry(parent, files[c]);
// 触发onDirectoryCreate/onFileCreate
doCreate(current[c]);
}
parent.setChildren(current);
}
- 对比规则 NameFileComparator
-
NAME_SYSTEM_COMPARATOR默认值，根据系统规则，Windows不区分大小写，Unix区分大小写 -
NAME_INSENSITIVE_COMPARATOR不区分大小写 -
NAME_COMPARATOR区分大小写
-
protected FileAlterationObserver(final FileEntry rootEntry, final FileFilter fileFilter,
final IOCase caseSensitivity) {
...
if (caseSensitivity == null || caseSensitivity.equals(IOCase.SYSTEM)) {
this.comparator = NameFileComparator.NAME_SYSTEM_COMPARATOR;
} else if (caseSensitivity.equals(IOCase.INSENSITIVE)) {
this.comparator = NameFileComparator.NAME_INSENSITIVE_COMPARATOR;
} else {
this.comparator = NameFileComparator.NAME_COMPARATOR;
}
}
依赖
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.7</version>
</dependency>
使用
- 定义监听器
public class ReportAlterationListener extends FileAlterationListenerAdaptor {
@Override
public void onStart(FileAlterationObserver observer) {
super.onStart(observer);
}
@Override
public void onStop(FileAlterationObserver observer) {
super.onStop(observer);
}
@Override
public void onFileCreate(File file) {
// upload to oss
}
}
- 启动监控器
public void start() {
// 设置监控目录和规则
FileAlterationObserver observer = new FileAlterationObserver(
"/Users/xxx/",
FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter(".pdf")));
observer.addListener(new ReportAlterationListener());
// 默认10000ms
FileAlterationMonitor monitor = new FileAlterationMonitor(3000, observer);
monitor.start();
}
其他
- 程序初始化时，需要执行一次全量的同步
- 程序宕机时，需要和OSS上报告进行对比，找出遗漏的报告，并上传
网友评论