监听FTP文件变化

整个项目的数据入口是监听FTP文件变化实现的。程序通过监听FTP目录内文件的变化(查找那些文件修改日期比上次修改日期晚的，然后在数据库中没有记录的文件)，把监听到新增的文件信息保存在数据库表内文件队列表，然后文件校验拓扑从文件队列表获取那些需要处理的文件信息然后进行拓扑处理。

配置文件

ZOOKEEPER_HOST_PORT=10.230.16.71:2181
ZOOKEEPER_HOST=10.230.16.71
ZOOKEEPER_PORT=2181
SERIAL_VERSION=20170922
#redis  IP
REDIS_HOST=10.230.16.77
REDIS_PORT=6379
#程序运行所在的机器IP
LOCAL_IP=10.230.16.77

监听程序入口

package com.lancy.main;

import java.util.List;
import java.util.Properties;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lancy.entity.FTPEntity;
import com.lancy.thread.GlobalFTPListener;
import com.lancy.thread.LocalFTPListener;

/**
* FTPListenMain FTP监听启动类 ,通过参数控制启动不同的监听
*/
public class FTPListenMain {
private static Logger logger = LoggerFactory.getLogger(FTPListenMain.class);
private static Properties props = new Properties();
private static final String profile = "config.properties";
static {
try {
props.load(FTPListenMain.class.getResourceAsStream("/" + profile));
} catch (Exception e) {
logger.error(e.getMessage());
}
}

//根据zookeeper里初始化的ftp配置信息启动对应的主机FTP的监听程序。zookeeper每配置一个ftp的主机信息，就启动一个监听线程
public static void main(String[] args) {
String zk_Host_Port = props.getProperty("ZOOKEEPER_HOST_PORT");
String local_Ip = props.getProperty("LOCAL_IP");
String redis_Host = props.getProperty("REDIS_HOST");
int redis_Port = Integer.parseInt(props.getProperty("REDIS_PORT"));
try {
// 参数为为空 不启动
if (args == null || args.length <= 0) {
logger.info(">>>>输入参数不正确>>>为空>>>");
return;
}
ZooKeeper zooKeeper = new ZooKeeper(zk_Host_Port, 1000, null);
if (args.length == 1 && "lnt".equals(args[0])) {
List<String> listPath = zooKeeper.getChildren("/lnt/ftp", false);
//获取zookeeper路径/lnt/ftp/下的所有节点,一般是ip地址
for (String nodePath : listPath) {
//nodeData是一个json字符串
//"{'host':'10.230.16.77','port':21,'username':'lnt','password':'123456','default':0,
'inPath':'input','outPath':'output'}"
String nodeData = new String(zooKeeper.getData("/lnt/ftp" + "/" + nodePath, false, null));

JSONObject json = JSON.parseObject(nodeData);
FTPEntity ftpEntity = new FTPEntity(json);
// 开启监听线程
if (!ftpEntity.getHost().equals("10.230.16.202")) {
logger.info(">>>>>>>>>>>>>>>>>>>开启监听线程,监听：" + ftpEntity.getHost() + ">>>>>>>>>>>>>>>>>>>>");
Thread listenLocalFTP = new Thread(new LocalFTPListener(ftpEntity, local_Ip, redis_Host, redis_Port));
listenLocalFTP.start();
}
}
} else if (args.length == 1 && "ykt".equals(args[0])) {
//差不多代码
} else if (args.length == 2 && "lnt".equals(args[0]) && "ykt".equals(args[1])) {
//上面两个代码相加
} else {
logger.info(">>>>输入参数不正确>>>>>>");
return;
}
zooKeeper.close();
} catch (Exception e) {
logger.info(e.getMessage());
e.printStackTrace();
}

}

}

监听线程类

package com.lancy.thread;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import org.apache.commons.net.ftp.FTPFile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSONObject;
import com.lancy.common.DButil;
import com.lancy.common.DateUtil;
import com.lancy.common.FTPClientUtil;
import com.lancy.entity.FTPEntity;

/**
* LocalFTPListener
*/
public class LocalFTPListener implements Runnable {
<
4000
span class="hljs-keyword">private static Logger logger = LoggerFactory.getLogger(LocalFTPListener.class);
private FTPEntity ftpEntity;
private String localIp;
private String redisHost;
private int redisPort;

/**
* 构造函数
*
* @param ftpEntity
* @param localIp
* @param redisHost
* @param redisPort
*/
public LocalFTPListener(FTPEntity ftpEntity, String localIp, String redisHost, int redisPort) {
this.ftpEntity = ftpEntity;
this.localIp = localIp;
this.redisHost = redisHost;
this.redisPort = redisPort;
}

@Override
public void run() {
try {
SimpleDateFormat dateFormate = new SimpleDateFormat("yyyyMMddHHmm");
// 监听所有的文件队列
Map<String, JSONObject> fileMap = ftpEntity.getFileMap();
boolean reConnect = true;
long reConnectTimeMillis = 0;// 重连时间
long startTimeMillis = 0;// 开始监听时间
long prevTimeMillis = 0;// 上次监听时间
long sysTimeMillis = 0;// 当前系统时间
FTPClientUtil client = new FTPClientUtil(ftpEntity.getHost(), ftpEntity.getPort(), ftpEntity.getUserName(), ftpEntity.getPassword());
logger.info("***connection to FTP:" + ftpEntity.getHost() + "/" + ftpEntity.getPort() + "*****");
logger.info("***userName:" + ftpEntity.getUserName() + ":" + ftpEntity.getPassword() + "*****");
//如果远程FTP连接成功
if (client.openConnection()) {
logger.info(">>>>>>>>>>>>连接" + ftpEntity.getHost() + "FTP服务器成功>>>>>>>>>>>>>>>>>>>");
// 获取最开始的扫描开始时间，当前系统时间精确到分钟的时间戳
startTimeMillis = System.currentTimeMillis();
prevTimeMillis = startTimeMillis;
while (true) {
try {
// 获取监听根目录下及其子目录下所有的文件集合
// 获取当天的文件包最后上传时间的字符串 例如：201706301600
//getEndTimeMillis方法里去连接redis，取redis里保存的最后上传时间
String endTime = DateUtil.getEndTimeMillis(new Date(), redisHost, redisPort);
logger.info(">>>>>>>>>>>>>>>>>>>[" + ftpEntity.getHost() + "]文件容器中文件数量：" + fileMap.size() + ">>>>>>>>>>>>>>>>>");
logger.info(">>>>>>>>>>>>>>>>>>>获取FTP[" + ftpEntity.getHost() + "]根目录下所有的一级目录>>>>>>>>>>>>>>>>>");
//获取给定目录(这里指根目录)下所有的文件夹，除了.和..之外的
List<String> firstDirectory = client.getFirstDirectory("");
List<FTPFile> filesAll = new ArrayList<>();
if (reConnectTimeMillis != 0) {
sysTimeMillis = System.currentTimeMillis();
logger.info("》》》》》》》》》重连的监听时间区间:" + dateFormate.format(new Date(reConnectTimeMillis)) + "--" + dateFormate.format(new Date(sysTimeMillis)) + "扫描FTP服务器[" + ftpEntity.getHost() + "]一级目录下的文件变化》》》》》》");
//遍历根目录下所有文件夹
for (String dir : firstDirectory) {
List<FTPFile> files = null;
if ("hmddc".equals(dir)) {
//获取目录及其子目录下新增的文件,并且文件的后缀符合指定文件后缀,文件的修改时间的时间戳大于等于prevTimeMillis(文件最后处理时间)
files = client.getPathFileAndChildPathFileElse(dir, reConnectTimeMillis, sysTimeMillis, "ZIP");
} else {
files = client.getPathFileAndChildPathFileElse(dir + "/input", reConnectTimeMillis, sysTimeMillis, "ZIP");
}
if (files != null && files.size() > 0) {
filesAll.addAll(files);
}
}
prevTimeMillis = sysTimeMillis;
reConnectTimeMillis = 0;
logger.info("》》》》》》》》》重连后,[" + ftpEntity.getHost() + "]服务器下的一级目录扫描完毕》》》》》》》》》》》》》");
} else {
sysTimeMillis = System.currentTimeMillis();
logger.info(">>>>>>>>>监听时间区间:" + dateFormate.format(new Date(prevTimeMillis - 1 * 60 * 1000)) + "--" + dateFormate.format(new Date(sysTimeMillis)) + "扫描FTP服务器[" + ftpEntity.getHost() + "]一级目录下的文件变化>>>>>>");
for (String dir : firstDirectory) {
List<FTPFile> files = null;
if ("hmddc".equals(dir)) {
files = client.getPathFileAndChildPathFileElse(dir, prevTimeMillis, sysTimeMillis, "ZIP");
} else {
files = client.getPathFileAndChildPathFileElse(dir + "/input", prevTimeMillis, sysTimeMillis, "ZIP");
}
if (files != null && files.size() > 0) {
filesAll.addAll(files);
}
}
prevTimeMillis = sysTimeMillis;
logger.info(">>>>>>>>>>>>[" + ftpEntity.getHost() + "]服务器下的一级目录扫描完毕>>>>>>>>>>>");
}

if (filesAll != null && filesAll.size() > 0) {
// 判断是否有新文件增加
for (FTPFile ftpFile : filesAll) {
// 记录当前文件的信息
String fileName = ftpFile.getName();
// 文件大小
String fileSize = String.valueOf(ftpFile.getSize());
// 文件的路径
String filePath = ftpFile.getLink();
// 文件的最后修改时间
String fileTime = dateFormate.format((ftpFile.getTimestamp().getTime()));
// 判断下这个文件是否已经加入文件监控和文件队列，只有加入的才能插入到文件队列表，供文件处理拓扑进行处理。
if (!fileMap.containsKey(filePath)) {
logger.info(">>>>>>>>>>>>>>>>>>>>>FTP服务器[" + ftpEntity.getHost() + "]有新文件增加:" + filePath + ">>>>>>>>>>>>>>>>>");
JSONObject jsonObject = new JSONObject();
jsonObject.put("fileName", fileName);
jsonObject.put("fileSize", fileSize);
jsonObject.put("filePath", filePath);
jsonObject.put("fileTime", fileTime);
// 加入文件队列中
fileMap.put(filePath, jsonObject);
// 添加文件信息到数据库
DButil.addFileMonitorInfor(localIp, ftpEntity.getHost(), jsonObject);
} else {
JSONObject fileJson = fileMap.get(filePath);
String preFileName = fileJson.getString("fileName");
String preFileSize = fileJson.getString("fileSize");
String preFilePath = fileJson.getString("filePath");
String preFileTime = fileJson.getString("fileTime");
if (fileName.equals(preFileName) && fileSize.equals(preFileSize) && filePath.equals(preFilePath) && fileTime.equals(preFileTime)) {
// 暂时默认上传完毕,一分钟内没变化，（上传完毕或者传送中）
JSONObject fileJsonEnd = fileMap.get(filePath);
// 加入文件队列
logger.info(">>>>>>>>>>>>>>>>>>>>>文件上传完毕,加入文件队列:" + preFileName + ">>>>>>>>>>>>>>>>>");
DButil.addFileQueue(ftpEntity.getHost(), fileJsonEnd, redisHost, redisPort, endTime);
// 移除
fileMap.remove(filePath);
} else {
// 表示还在继续传，修改队列中的信息，继续检查是否上传完毕
logger.info(">>>>>>>>>>>>>>>>>>>>>FTP服务器：" + ftpEntity.getHost() + "有文件正在上传中>>>>>>>>>>>>>>>>>");
fileJson.put("fileName", fileName);
fileJson.put("fileSize", fileSize);
fileJson.put("filePath", filePath);
fileJson.put("fileTime", fileTime);
fileMap.put(filePath, fileJson);
DButil.addFileMonitorInfor(localIp, ftpEntity.getHost(), fileJson);
}
}
}
} else {
// 当监听没有新文件增加，旧文件发生改变的时候
if (fileMap != null && fileMap.size() > 0) {
// 加入文件队列
logger.info(">>>>>>>>>>>>>>>[" + ftpEntity.getHost() + "]暂时没有新文件增加,清空文件容器中的存在的文件>>>>>>>>>>>>>>>>>>>");
for (String filePath : fileMap.keySet()) {
JSONObject fileJson = fileMap.get(filePath);
DButil.addFileQueue(ftpEntity.getHost(), fileJson, redisHost, redisPort, endTime);
// 移除
fileMap.remove(filePath);
}
}
}

} catch (Exception e) {
// 捕获FTP连接服务器过程中出现的异常
logger.error("》》》》》》》》》》》》》[" + ftpEntity.getHost() + "]" + e.getMessage() + "》》》》》》》》》》》》》》》》");
e.printStackTrace();
reConnectTimeMillis = prevTimeMillis;
int count = 0;
while (true) {
count++;
logger.info("》》》》》》》》》》》》》1分钟后尝试重连[" + ftpEntity.getHost() + "]》》》》》》》》》》》》》》》》》");
Thread.sleep(1 * 60 * 1000);
logger.info("》》》》》》》》》》》》》开始重连[" + ftpEntity.getHost() + "]》》》》》》》》》》》》》》》》》》》");
reConnect = client.checkAndReconnect(ftpEntity.getHost(), ftpEntity.getPort(), ftpEntity.getUserName(), ftpEntity.getPassword());
logger.info("》》》》》》》》》》》》》》第[" + count + "]次重连[" + ftpEntity.getHost() + "]" + (reConnect == true ? "成功" : "失败") + "》》》》》》》》》》》》》》》》》");
if (reConnect) {
break;
}

}
}
if (reConnectTimeMillis != 0) {
prevTimeMillis = reConnectTimeMillis;
logger.info("》》》》》》》》》》》》》》》》》》》》》》2分钟再次扫描从断点:" + dateFormate.format(new Date(reConnectTimeMillis)) + "开始扫描[" + ftpEntity.getHost() + "]FTP服务器》》》》》》》》》》》》》》》》");
Thread.sleep(1000 * 60 * 2);
} else {
logger.info(">>>>>>>>>>>>>>>>>>>>>2分钟再次扫描[" + ftpEntity.getHost() + "]FTP服务器>>>>>>>>>>>>>>>>>");
Thread.sleep(1000 * 60 * 2);
}

}
} else {
logger.info(">>>>>>>>>>>>>>>>>>>>>连接不上FTP>>>>>>>>>>>>>>>>>>>>>>>>");
}
} catch (Exception e) {
logger.error(">>>>>>>>>>>>>>>>" + e.getMessage() + ">>>>>>>>>>>>>>>>>>");
e.printStackTrace();
}

}

public String getLocalIp() {
return localIp;
}

public void setLocalIp(String localIp) {
this.localIp = localIp;
}

public FTPEntity getEntity() {
return ftpEntity;
}

public void setEntity(FTPEntity ftpEntity) {
this.ftpEntity = ftpEntity;
}

public String getRedisHost() {
return redisHost;
}

public void setRedisHost(String redisHost) {
this.redisHost = redisHost;
}

public int getRedisPort() {
return redisPort;
}

public void setRedisPort(int redisPort) {
this.redisPort = redisPort;
}

public static void main(String[] args) {
String s = "sztckp/input/CZ2132000320161120.ZIP";
System.out.println(s.substring(0, s.indexOf("/")));
}
}

FTP信息实体

package com.lancy.entity;

import com.alibaba.fastjson.JSONObject;
import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
* FTPEntity FTP连接信息实体类
*/
@SuppressWarnings("serial")
public class FTPEntity implements Serializable {
private String host;
private int port;
private String userName;
private String password;
private int isDefault;
private String inPath;// 所监控的根目录
private String outPath;// 下发的根目录
private Map<String, JSONObject> fileMap;// 目的FTP监控根目录下所有的文件Map

public FTPEntity() {
}

/**
* 构造函数
*
* @param json
*            FTP信息
*/
public FTPEntity(JSONObject json) {
this.host = json.getString("host");
this.port = json.getInteger("port");
this.userName = json.getString("username");
this.password = json.getString("password");
this.isDefault = json.getIntValue("default");
this.inPath = json.getString("inPath");
this.outPath = json.getString("outPath");
this.fileMap = new ConcurrentHashMap<String, JSONObject>();
}

public String getHost() {
return host;
}

public void setHost(String host) {
this.host = host;
}

public int getPort() {
return port;
}

public void setPort(int port) {
this.port = port;
}

public String getUserName() {
return userName;
}

public void setUserName(String userName) {
this.userName = userName;
}

public String getPassword() {
return password;
}

public void setPassword(String password) {
this.password = password;
}

public int getIsDefaule() {
return isDefault;
}

public void setIsDefaule(int isDefault) {
this.isDefault = isDefault;
}

public String getOutPath() {
return outPath;
}

public void setOutPath(String outPath) {
this.outPath = outPath;
}

public String getInPath() {
return inPath;
}

public void setInPath(String inPath) {
this.inPath = inPath;
}

public int getIsDefault() {
return isDefault;
}

public void setIsDefault(int isDefault) {
this.isDefault = isDefault;
}

public Map<String, JSONObject> getFileMap() {
return fileMap;
}

public void setFileMap(Map<String, JSONObject> fileMap) {
this.fileMap = fileMap;
}

@Override
public String toString() {
StringBuilder builder = new StringBuilder("{");
builder.append("host:").append(this.host).append(",");
builder.append("port:").append(this.port).append(",");
builder.append("userName:").append(this.userName).append(",");
builder.append("password:").append(this.password).append(",");
builder.append("default:").append(this.isDefault).append(",");
builder.append("inPath:").append(this.inPath).append(",");
builder.append("outPath:").append(this.outPath);
builder.append("fileMap:").append(this.fileMap);
builder.append("}");
return builder.toString();
}
}