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服务的过程，我这演示了匿名创建站点，谁都可以访问，还可以新建用户，需要用户登录才能访问。 源码 引入该依赖 commons-net commons-net 3.6 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 文件发生了改变 * */ @Function void 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 { @Autowired private FTPConfig ftpConfig; private String SPLIT = ":"; private ThreadLocal currentFTPClient; private ThreadLocal currentListener; public FTPServiceImpl() { this.currentFTPClient = new ThreadLocal<>(); this.currentListener = new ThreadLocal<>(); } @Override public 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; } } @Override public boolean loginOut() { try { currentFTPClient.get().logout(); currentFTPClient.get().disconnect(); return Boolean.TRUE; } catch (Exception e) { return Boolean.FALSE; } } @Override public FTPFile[] listFile() { FTPClient ftpClient = this.currentFTPClient.get(); try { return ftpClient.listFiles(); } catch (Exception e) { return null; } } @Override public 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 fileMemory; private final FileChangeEvent fileChangeEvent; public ListenerFileChangeThreadRunnable(FTPClient ftpClient, FileChangeEvent fileChangeEvent) { this.ftpClient = ftpClient; this.fileChangeEvent = fileChangeEvent; this.fileMemory = new HashMap<>(); } @Override public void run() { while (!stop) { try { FTPFile[] ftpFiles = ftpClient.listFiles(); //判断文件被删除 if (fileMemory.size() > 0) { Set fileNames = new HashSet<>(); for (FTPFile ftpFile : ftpFiles) { if (ftpFile.isDirectory()) { log.info("文件夹不做删除判断"); continue; } fileNames.add(ftpFile.getName()); } Set> entries = fileMemory.entrySet(); for (Map.Entry 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); } } } @Override public 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, "文件删除"); @Getter private Integer type; @Getter private 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 { @Autowired private 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 listStr = new ArrayList<>();//读取的数据在listStr while ((s = reader.readLine()) != null) { System.out.println("===================>" + s); listStr.add(s); } //处理业务逻辑 inputStream.close(); reader.close(); ftpClient.completePendingCommand(); } }); } @Override public void run(ApplicationArguments args) throws Exception { ftpTest(); } }