基于ftp协议的文件变化主动监听

前言

文件传输协议（File Transfer Protocol，FTP）是用于在网络上进行文件传输的一套标准协议，它工作在 OSI 模型的第七层， TCP 模型的第四层， 即应用层， 使用 TCP 传输而不是 UDP， 客户在和服务器建立连接前要经过一个“三次握手”的过程， 保证客户与服务器之间的连接是可靠的， 而且是面向连接， 为数据传输提供可靠保证。以上是百度百科的官方解释，通俗点讲就是ftp是通过一种协议让你能够远程访问其他计算机上的文件数据，虽然ftp能够访问到远程服务器的文件，甚至可以下载远程服务器的文件，但是ftp没有被动监听功能，无法监听到远程服务器的文件变化，这时侯就需要我们在客户端去做主动监听。

实现思路

其实主动监听比较简单，主动监听的大致思路就是轮询，没错就是轮询很通俗易懂的一种思路。通过ftp协议连接到远程服务器，之后通过循环不断的执行查看文件列表操作，通过第一次查找的缓存目录去比对后续的查找结果；如果出现后一次文件名称在缓存列表中不存在则证明该文件是新增，如果出现后一次文件名称在缓存列表中存在且文件大小不同，则证明文件被更改，反之则证明该文件无变化；如果出现缓存列表中文件存在但是后一次文件列表中不存在的文件名称，则名称该文件是删除；如上变化除无变化状态以外其余都需要更改本地缓存，以保证每次对比的准确性；想必看完如上设计大家也会发现一些弊端，比如我一个文本文件原文件内容是abc，但是我后来更改成了abd，该种情况根据如上设计方式则无法监听到文件的变化，如果要实现较为精准的变化监听则最好将远程文件缓存到本地之后利用文件md5值的方式来判断，由于本文所述的监听方式主要是类似于系统日志文件这种不经常变化文件名称和篡改文件已有内容的场景，所以利用md5值比对的方式本文不会介绍及实现。

代码实现思路

代码设计中采用事件的方式，利用监听的形式来进行文件夹内容的监听，设计类有如下ListenerFileChangeThreadRunnable（文件变化监听线程类）、ListenerChangeRunnable（文件变化线程接口）、FileChangeType（文件变化类型枚举）、FileChangeEvent（文件变化事件接口）、FileChangeData（文件变化类）、FTPServiceImpl（FTP服务类）。
具体思路为利用commons-net软件包中提供的ftp相关的类进行ftp通信操作，通过监听时建立ListenerChangeRunnable线程来论询远程服务文件列表，当监听到实现思路中的变化时利用FileChangeEvent接口发送文件变化事件，并通过FileChangeType枚举类标记文件变化类型。

具体代码实现

依赖引入

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

FTPService接口

public interface FTPService {

    /**
     * ftp登陆
     * @return boolean 是否登陆成功
     * */
    boolean login();

    /**
     * ftp登出
     * @return boolean 是否登出成功
     * */
    boolean loginOut();

    /**
     * 获取文件列表
     * @return FTPFile[] 文件列表
     * */
    FTPFile[] listFile();

    /**
     * 监听文件夹的改变
     * @param fileChangeEvent 文件改变事件
     * */
    void addListenerFileChange(FileChangeEvent fileChangeEvent);
}

FTPServiceImpl类

@Service
public class FTPServiceImpl implements FTPService {

    @Autowired
    private FTPConfig ftpConfig;

    private String SPLIT = ":";

    private ThreadLocal<FTPClient> currentFTPClient;

    private ThreadLocal<ListenerChangeRunnable> currentListener;

    public FTPServiceImpl() {
        this.currentFTPClient = new ThreadLocal<>();
        this.currentListener = new ThreadLocal<>();
    }

    @Override
    public boolean login() {
        FTPClient ftpClient = new FTPClient();
        try {
            ftpClient.connect(ftpConfig.getFtpIp(), ftpConfig.getFtpPort());
            ftpClient.login(ftpConfig.getUsername(), ftpConfig.getPassword());
            ftpClient.setControlEncoding("gb2312");

            ftpClient.changeWorkingDirectory(new String(ftpConfig.getWorkspace().getBytes("GBK"), "iso-8859-1"));
            this.currentFTPClient.set(ftpClient);
            return Boolean.TRUE;
        } catch (Exception e) {
            return Boolean.FALSE;
        }
    }

    @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();
    }
}

FileChangeEvent接口

public interface FileChangeEvent {

    /**
     * 文件发生改变时触发此方法
     * @param fileChangeData 文件发生了改变
     * */
    @Function
    void change(FileChangeData fileChangeData);
}

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() {
        //如果是删除事件则不能够获取流
        if (Objects.equals(eventType, FileChangeType.FILE_DELETED)) {
            return null;
        }

        try {
            return ftpClient.retrieveFileStream(this.fileName);
        } catch (IOException e) {
            return null;
        }
    }
}

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;
    }
}

ListenerChangeRunnable枚举

public interface ListenerChangeRunnable extends Runnable {

    /**
     * 停止监听文件
     * @return boolean 是否停止成功
     * */
    boolean stopListener();
}

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<>();
    }

    @Override
    public 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());
                            fileChangeEvent.change(fileChangeData);
                        }
                    }
                }
                //判断文件是否有更改或新增
                for (FTPFile ftpFile: ftpFiles) {
                    //判断是否为文件夹
                    if (ftpFile.isDirectory()) {
//                        log.info("{}为文件不进行监听操作", ftpFile.getName());
                        continue;
                    }
                    FileChangeData fileChangeData = new FileChangeData();
                    fileChangeData.setFileName(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) {
                throw new RuntimeException(e);
            }
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public boolean stopListener() {
        this.stop = Boolean.TRUE;
        this.fileMemory.clear();
        return this.stop;
    }
}

FTPConfig配置类

@Data
@Configuration
public class FTPConfig {

    @Value("${ftp.ip:127.0.0.1}")
    private String ftpIp;

    @Value("${ftp.port:21}")
    private Integer ftpPort;

    @Value("${ftp.username:root}")
    private String username;

    @Value("${ftp.password:root}")
    private String password;

    @Value("${ftp.workspace:root}")
    private String workspace;
}

使用举例

@SpringBootTest
class SendEmailApplicationTests {
   @Autowired
   private FTPService ftpService;
   @Test
   void ftpTest() {
        ftpService.login();
        FTPFile[] ftpFiles = ftpService.listFile();
        for (FTPFile file : ftpFiles) {
            System.out.println(String.format("filename:%s,filesize:%s", file.getName(), file.getSize()));
        }
        ftpService.addListenerFileChange(ftpFile -> {
            System.out.println(String.format("文件%s被改变了，文件改变类型%s", ftpFile.getFileName(), ftpFile.getEventType().getDesc()));
        });
    } 
}

结语

如上只是面对ftp文件变化监听问题的一种实现思路及具体实现代码，但并非最优解，如关于本篇内容存在疑问欢迎评论提问或者私聊，同时如果大家对于该种场景需求较多作者可以将该部分内容封装为starter供大家学习使用。