java通过FTP跨服务器动态监听读取指定目录下文件数据

背景：

1、文件数据在A服务器（windows）（不定期在指定目录下生成），项目应用部署在B服务器（Linux）；
2、项目应用在B服务器，监听A服务器指定目录，有新生成文件，进行读取文件信息，持久化数据；
3、提供两块内容，第一安装windows FTP服务；第二项目源码，希望可以帮助到你。

共计4种方案，试错采用了第三种方案，第四种方案没有试。

1、使用jcsh.jar提供方法读取文件信息，但需要A服务器开通SSH远程连接，一般linux服务器都是默认开通的，可直接读取连接读取，windows系统需安装SSH，因现场环境A服务器是windows2003，故放弃这种方法。
2、曲线救国，通过脚本（脚本监听比较困难，故放弃）把A服务器信息定时存入B服务器（Linux），再通过jcsh.jar读取文件信息。
3、通过A服务器安装FTP服务，B服务器安装FTP客户端，使用java动态监听该目录下生成文件读取信息。
4、把A服务器指定目录进行共享（等同于共享的这个目录就是B服务的目录了），再进行读取，因第三种方案成功，故没有尝试第四种方案。

windows安装FTP服务

1、开启ftp服务：控制面板–程序和功能–启用或关闭windows功能–标红框全部打开–点击确定

2、新建站点：
控制面板–大图标–管理工具

IIS管理器

网站–添加FTP站点




以上就是windows安装FTP服务的过程，我这演示了匿名创建站点，谁都可以访问，还可以新建用户，需要用户登录才能访问。

源码

引入该依赖

<dependency><groupId>commons-net</groupId><artifactId>commons-net</artifactId><version>3.6</version>
</dependency>

FileChangeData

@Data
public class FileChangeData {/*** 文件信息* */private FTPFile ftpFile;/*** 文件改变类型* */private FileChangeType eventType;/*** 文件名称* */private   String fileName;/*** 文件大小* */private Long fileSize;/*** FTPClient* */private FTPClient ftpClient;/*** 获取文件输入流* @return InputStream* */public InputStream getInputStream(String filePathName) {//如果是删除事件则不能够获取流if (Objects.equals(eventType, FileChangeType.FILE_DELETED)) {return null;}try {return ftpClient.retrieveFileStream(filePathName);} catch (IOException e) {return null;}}
}

FileChangeEvent

public interface FileChangeEvent {/*** 文件发生改变时触发此方法* @param fileChangeData 文件发生了改变* */@Functionvoid change(FileChangeData fileChangeData) throws IOException;
}

FTPService

public interface FTPService {/*** ftp登陆* @return boolean 是否登陆成功* */FTPClient login();/*** ftp登出* @return boolean 是否登出成功* */boolean loginOut();/*** 获取文件列表* @return FTPFile[] 文件列表* */FTPFile[] listFile();/*** 监听文件夹的改变* @param fileChangeEvent 文件改变事件* */void addListenerFileChange(FileChangeEvent fileChangeEvent);
}

ListenerChangeRunnable

public interface ListenerChangeRunnable extends Runnable {/*** 停止监听文件* @return boolean 是否停止成功* */boolean stopListener();
}

FTPServiceImpl

@Service
public class FTPServiceImpl implements FTPService {@Autowiredprivate FTPConfig ftpConfig;private String SPLIT = ":";private ThreadLocal<FTPClient> currentFTPClient;private ThreadLocal<ListenerChangeRunnable> currentListener;public FTPServiceImpl() {this.currentFTPClient = new ThreadLocal<>();this.currentListener = new ThreadLocal<>();}@Overridepublic FTPClient login() {FTPClient ftpClient = new FTPClient();try {ftpClient.connect(ftpConfig.getFtpIp(), ftpConfig.getFtpPort());ftpClient.login(ftpConfig.getUsername(), ftpConfig.getPassword());
//            ftpClient.setControlEncoding("gb2312");this.currentFTPClient.set(ftpClient);return ftpClient;} catch (Exception e) {return null;}}@Overridepublic boolean loginOut() {try {currentFTPClient.get().logout();currentFTPClient.get().disconnect();return Boolean.TRUE;} catch (Exception e) {return Boolean.FALSE;}}@Overridepublic FTPFile[] listFile() {FTPClient ftpClient = this.currentFTPClient.get();try {return ftpClient.listFiles();} catch (Exception e) {return null;}}@Overridepublic void addListenerFileChange(FileChangeEvent fileChangeEvent) {FTPClient ftpClient = this.currentFTPClient.get();ListenerFileChangeThreadRunnable listenerFileChangeThread = new ListenerFileChangeThreadRunnable(ftpClient, fileChangeEvent);this.currentListener.set(listenerFileChangeThread);new Thread(listenerFileChangeThread).start();}
}

ListenerFileChangeThreadRunnable

@Slf4j
public class ListenerFileChangeThreadRunnable implements ListenerChangeRunnable {private final FTPClient ftpClient;private volatile boolean stop;private final Map<String, Long> fileMemory;private final FileChangeEvent fileChangeEvent;public ListenerFileChangeThreadRunnable(FTPClient ftpClient, FileChangeEvent fileChangeEvent) {this.ftpClient = ftpClient;this.fileChangeEvent = fileChangeEvent;this.fileMemory = new HashMap<>();}@Overridepublic void run() {while (!stop) {try {FTPFile[] ftpFiles = ftpClient.listFiles();//判断文件被删除if (fileMemory.size() > 0) {Set<String> fileNames = new HashSet<>();for (FTPFile ftpFile : ftpFiles) {if (ftpFile.isDirectory()) {log.info("文件夹不做删除判断");continue;}fileNames.add(ftpFile.getName());}Set<Map.Entry<String, Long>> entries = fileMemory.entrySet();for (Map.Entry<String, Long> map : entries) {if (!fileNames.contains(map.getKey())) {log.info("文件{}被删除了", map.getKey());FileChangeData fileChangeData = new FileChangeData();fileChangeData.setEventType(FileChangeType.FILE_DELETED);fileChangeData.setFileName(map.getKey());fileChangeData.setFileSize(map.getValue());fileMemory.remove(map.getKey());entries.remove(map.getKey());fileChangeEvent.change(fileChangeData);}}}//判断文件是否有更改或新增for (FTPFile ftpFile: ftpFiles) {//判断是否为文件夹if (ftpFile.isDirectory()) {
//                        log.info("{}为文件不进行监听操作", ftpFile.getName());continue;}FileChangeData fileChangeData = new FileChangeData();fileChangeData.setFileName(ftpFile.getName());
//                    fileChangeData.setFileName("D:\\ftptest\\aaa\\"+ftpFile.getName());fileChangeData.setFileSize(ftpFile.getSize());fileChangeData.setFtpFile(ftpFile);//文件是否存在于缓存文件列表中if (fileMemory.containsKey(ftpFile.getName())) {
//                        log.info("文件{}在内存中已经存在，进行大小判断", ftpFile.getName());if (!Objects.equals(fileMemory.get(ftpFile.getName()), ftpFile.getSize())) {
//                            log.info("文件{}在内存中已经存在且大小不一致，进行更新缓存操作", ftpFile.getName());fileMemory.put(ftpFile.getName(), ftpFile.getSize());fileChangeData.setEventType(FileChangeType.FILE_UPDATE);fileChangeEvent.change(fileChangeData);}continue;}
//                    log.info("文件{}在内存中不存在进行缓存操作", ftpFile.getName());fileMemory.put(ftpFile.getName(), ftpFile.getSize());fileChangeData.setEventType(FileChangeType.FILE_ADD);fileChangeEvent.change(fileChangeData);}} catch (Exception e) {continue;//throw new RuntimeException(e);}try {TimeUnit.SECONDS.sleep(20);} catch (InterruptedException e) {continue;//throw new RuntimeException(e);}}}@Overridepublic boolean stopListener() {this.stop = Boolean.TRUE;this.fileMemory.clear();return this.stop;}
}

FileChangeType

public enum FileChangeType {FILE_UPDATE(0, "文件更新"),FILE_ADD(1, "文件添加"),FILE_DELETED(2, "文件删除");@Getterprivate Integer type;@Getterprivate String desc;FileChangeType(Integer type, String desc) {this.type = type;this.desc = desc;}
}

FTPConfig

@Data
@Configuration
public class FTPConfig {@Value("${ftp.ip:ftp的ip}")private String ftpIp;@Value("${ftp.port:ftp端口，默认21}")private Integer ftpPort;@Value("${ftp.username:ftp创建的用户名}")private String username;@Value("${ftp.password:ftp创建的用户名密码}")private String password;
}

SendEmailApplicationTests

@Component
class SendEmailApplicationTests implements ApplicationRunner {@Autowiredprivate FTPService ftpService;void ftpTest() {FTPClient ftpClient = ftpService.login();ftpService.addListenerFileChange(ftpFile -> {System.out.println(String.format("文件%s被改变了，文件改变类型%s", ftpFile.getFileName(), ftpFile.getEventType().getDesc()));InputStream inputStream = ftpClient.retrieveFileStream("/"+ ftpFile.getFileName());if(inputStream!=null){BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream,"GBK"));String s = null;List<String> listStr = new ArrayList<>();//读取的数据在listStrwhile ((s = reader.readLine()) != null) {System.out.println("===================>" + s);listStr.add(s);}//处理业务逻辑inputStream.close();reader.close();ftpClient.completePendingCommand();}});}@Overridepublic void run(ApplicationArguments args) throws Exception {ftpTest();}
}