FTP和SFTP介绍(常见的和java代码遇到的问题)

古城微笑少年丶 2022-12-22 10:00 140阅读 0赞

# 背景 #

最近做导数服务的时候需要支持FTP和SFTP，这块本来让我带的人做，正好我这块也没事，就写了下了解了解

# FTP #

连接命令上传下载不细讲 网上一堆

#### 主动模式和被动模式 ####

**PORT（主动）方式的连接过程是**：客户端向服务器的FTP端口（默认是21）发送连接请求，服务器接受连接，建立一条命令链路。当需要传送数据时， 客户端在命令链路上用PORT命令告诉服务器："我打开了XX端口，你过来连接我”。于是服务器从20端口向客户端的XX端口发送连接请求，建立 一条数据链路来传送数据。

**PASV（被动）方式的连接过程是**：客户端向服务器的FTP端口（默认是21）发送连接请求，服务器接受连接，建立一条命令链路。当需要传送数据时， 服务器在命令链路上用PASV命令告诉客户端：“我打开了XX端口，你过来连接我”。于是客户端向服务器的XX端口发送连接请求，建立一条数据链 路来传送数据。

java中，内网用被动模式，外网连接时用主动模式

java代码设置方式

> ftp.enterLocalPassiveMode();(被动)  
> ftp.enterLocalActiveMode();(主动)

### java拷贝数据上传 ###

##### ftp拷贝IO的时候遇到的问题 #####

我在递归拷贝sftp的文件到hdfs的时候，报错

> inputStream is null

的异常，原因是因为我们这里要在每次拷贝一个文件的时候（也就是每个io）需要将inputstream 和ftp都需要关闭才可以。

##### 代码(这里只展示一部分代码,我在最后的代码uploader.uploadFtpToHdfs(srcPath, dstPath, channelSftp);中关闭了sftp和inputStream) #####

public void uploadDataToHdfs(String srcFilePath, Path dstPath) throws Exception {
            HDFSInfo info = createHdfsInfo(hdfsInfo);
            try (FileSystem fs = info.getFileSystem();
                 ConcurrencyHdfsUploader uploader = new ConcurrencyHdfsUploader(fs, 5)) {
                FTPClient ftp = connectFtp(ftpInfo);
                if (!ftp.changeWorkingDirectory(srcFilePath)) {
                    String fileName = ftp.listFiles(srcFilePath)[0].getName();
                    InputStream inputStream = ftp.retrieveFileStream(srcFilePath);
                    Path resultPath = new Path(dstPath, fileName);
                    uploader.uploadStreamToHdfs(inputStream, resultPath, true);
                } else {
                    boolean recursion = false;
                    if (ftpInfo.isSetOptions()) {
                        if (ftpInfo.getOptions().get(OptionConstant.RECURSION) != null) {
                            if ("TRUE".equals(ftpInfo.getOptions().get(OptionConstant.RECURSION).toUpperCase())) {
                                recursion = true;
                            }
                        }
                    }
                    upload(ftp, srcFilePath, dstPath, uploader, recursion);
                }
                uploader.waitAllJobsDone();
            }
        }
    
        private void upload(FTPClient ftp, String srcFilePath, Path hdfsPath, ConcurrencyHdfsUploader uploader, boolean recursion) throws Exception {
            ftp.changeWorkingDirectory(srcFilePath);
            FTPFile[] ftpFiles = ftp.listFiles(srcFilePath);
            for (FTPFile file : ftpFiles) {
                Path dstPath = new Path(hdfsPath, file.getName());
                String srcPath = srcFilePath + File.separator + file.getName();
                if (file.isDirectory()) {
                    if (recursion) {
                        upload(ftp, srcPath, dstPath, uploader, recursion);
                    } else {
                        continue;
                    }
                } else {
                    FTPClient ftpClient = connectFtp(ftpInfo);
                    uploader.uploadFtpToHdfs(srcPath, dstPath, ftpClient);
                }
            }
            ftp.disconnect();
        }

# sftp #

sftp是一种安全传送协议，也是一个服务器，默认22端口。比如filezilla连接服务器的时候默认就是22端口连接的。

#### 常用命令 ####

连接  
sftp -P port username@IP

上传  
put \[本地文件的地址\] \[服务器上文件存储的位置\]

下载  
get \[服务器上文件存储的位置\] \[本地要存储的位置\]

#### java拷贝数据上传 ####

##### java sftp在io时候遇到的问题 #####

我在递归拷贝sftp的文件到hdfs的时候，报错

> sftp inputstream is closed

单个文件是能成功的。问题和ftp的原因相似，也是每次拷贝一个io流的时候，需要将inputStream和sftp关闭才可以

##### 代码(这里只展示一部分代码,我在最后的代码uploader.uploadSftpToHdfs(srcPath, dstPath, channelSftp);中关闭了sftp和inputStream) #####

public void uploadSftpToHDFS(String srcFilePath, Path hdfsPath) throws Exception {
            HDFSInfo hdfsInfo = createHdfsInfo(this.hdfsInfo);
            try (FileSystem fs = hdfsInfo.getFileSystem();
                 ConcurrencyHdfsUploader uploader = new ConcurrencyHdfsUploader(fs, 5)) {
                ChannelSftp channelSftp = connectSftp(sftpInfo);
                if (!channelSftp.stat(srcFilePath).isDir()) {
                    InputStream inputStream = channelSftp.get(srcFilePath);
                    String filename = ((ChannelSftp.LsEntry) channelSftp.ls(srcFilePath).get(0)).getFilename();
                    Path dstPath = new Path(hdfsPath, filename);
                    uploader.uploadStreamToHdfs(inputStream, dstPath, true);
                } else {
                    boolean recursion = false;
                    if (sftpInfo.isSetOptions()) {
                        if (sftpInfo.getOptions().get(OptionConstant.RECURSION) != null) {
                            if ("TRUE".equals(sftpInfo.getOptions().get(OptionConstant.RECURSION).toUpperCase())) {
                                recursion = true;
                            }
                        }
                    }
                    upload(channelSftp, srcFilePath, hdfsPath, uploader, recursion);
                }
                uploader.waitAllJobsDone();
                channelSftp.disconnect();
            }
        }
    
        public void upload(ChannelSftp sftp, String srcFilePath, Path hdfsPath,
                           ConcurrencyHdfsUploader uploader,
                           boolean recursion) throws Exception {
            sftp.cd(srcFilePath);
            Vector<ChannelSftp.LsEntry> ls = sftp.ls(srcFilePath);
            for (ChannelSftp.LsEntry entry : ls) {
                final SftpATTRS attrs = entry.getAttrs();
                String srcPath = sftp.pwd() + File.separator + entry.getFilename();
                Path dstPath = new Path(hdfsPath, entry.getFilename());
                if (attrs.isDir()) {
                    if (recursion) {
                        if (".".equals(entry.getFilename()) || "..".equals(entry.getFilename())) {
                            continue;
                        }
                        upload(sftp, srcPath, dstPath, uploader, recursion);
                    } else {
                        continue;
                    }
                } else {
                    ChannelSftp channelSftp = connectSftp(sftpInfo);
                    String filename = entry.getFilename();
                    dstPath = new Path(hdfsPath, filename);
                    uploader.uploadSftpToHdfs(srcPath, dstPath, channelSftp);
                }
            }
        }