近日有个需求,需要将远程SFTP服务器上的一个压缩包内的数据解析导入到我方数据库,压缩包内第一级目录为同名文件夹,第二级目录才是存放数据的txt文件。 参考文章:https://www.jianshu.com/p/79bfe182a28f
我想得到的是压缩包内的数据,并不想把这个压缩包下载到本地再解析,所以我直接通过sftp获取zip文件的字节输入流,并将该字节输入流作为构造参数构造出压缩输入流。 然后通过压缩输入流的getNextEntry方法,循环遍历该压缩流内部的压缩实体。最终实现代码如下: /**
* 从FTP服务器解析错误文件返回文件内容集合---外部方法调用本方法获得文件内容,解析成数据库对象并存入数据库
* @param ftpdir
* 要从FTP下载的文件夹路径
* @param downloadFileName
* 要从FTP下载的文件的文件名
* @throws Exception
*/
public Map<String,List<String>> analysisData(String ftpdir, String downloadFileName) throws Exception {
//获得缓冲输入流
BufferedInputStream bs = null;
//压缩实体
ZipEntry zipEntry = null;
//将每个压缩实体字节读入的字节数组
byte[] bytes = null;
//获得字符缓冲输入流
BufferedReader br = null;
//获取整个zip文件的输入流
InputStream inputStreamAll = null;
//将整个zip文件的输入流转为压缩输入流
ZipInputStream zipInputStream = null;
//获取单个压缩实体的输入流
InputStream inputStream=null;
//定义返回对象:key为压缩实体的文件名,value为压缩实体的内容集合
Map<String,List<String>> map = new HashMap<String,List<String>>();
try {
//进入要下载的文件路径
sftp.cd(ftpdir);
//获取整个zip文件的输入流
inputStreamAll = sftp.get(downloadFileName);
//将整个zip文件的输入流转为压缩输入流
zipInputStream = new ZipInputStream(inputStreamAll);
//当压缩文件内有文件时候进入解析
while ((zipEntry = zipInputStream.getNextEntry()) != null){
//不为文件夹时走如下逻辑
if (!zipEntry.isDirectory()) {
//当前压缩实体的文件名
String txtName = zipEntry.getName();
long size = zipEntry.getSize();
// ZipEntry的size可能为-1,表示未知
if (size == -1) {
//直接读取当前ZipEntry的流,直到为-1为止
ByteArrayOutputStream baos = new ByteArrayOutputStream();
while (true) {
int entryBytes = zipInputStream.read();
if (entryBytes == -1) break;
baos.write(entryBytes);
}
baos.close();
inputStream = new ByteArrayInputStream(baos.toByteArray());
} else { // ZipEntry的size正常
//将压缩实体通过缓冲输入流读入字节数组,然后将该字节数组转换成输入流
bytes = new byte[(int) size];
bs = new BufferedInputStream(zipInputStream);
bs.read(bytes, 0, (int) zipEntry.getSize());
inputStream = new ByteArrayInputStream(bytes);
}
//读入文件,编码格式utf-8
br = new BufferedReader(new InputStreamReader(inputStream,"UTF-8"));
//压缩实体的每行内容集合 -工具类方法
List<String> listTxt = IOUtils.readLines(br);
map.put(txtName, listTxt);
}
}
} catch (Exception e) {
logger.error("analysisData方法异常:",e);
throw e;
} finally {
if (bs != null) {
try {
bs.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
bs = null;
}
}else{
if(zipInputStream != null){
try {
zipInputStream.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
zipInputStream = null;
}
}
}
if (br != null) {
try {
br.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
br = null;
}
}
//断开FTP连接
disconnect();
}
return map;
}
因为需求相关,有时还需要往远程服务器上传或下载文件,附上创建FTP连接实例并进行各种简易交互的完整工具类: package com.aic.erecord.service.utils;
import java.io.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import org.apache.commons.io.IOUtils;
import org.apache.log4j.Logger;
import com.jcraft.jsch.*;
public class FtpJSch {
private static ChannelSftp sftp = null;
private static final Logger logger = Logger.getLogger(FtpJSch.class);
/**
* 获取FTP连接实例方法
* @param ftphost FTP服务器地址
* @param ftpusername FTP服务器用户名
* @param ftppassword FTP服务器密码
* @param ftpport FTP服务器端口
* @return FtpJSch对象
* @throws Exception
*/
public static FtpJSch getConnect(String ftphost, String ftpusername, String ftppassword, int ftpport) throws Exception {
FtpJSch ftp = new FtpJSch();
try {
JSch jsch = new JSch();
// 获取sshSession 账号-ip-端口
Session sshSession = jsch.getSession(ftpusername, ftphost, ftpport);
// 添加密码
sshSession.setPassword(ftppassword);
Properties sshConfig = new Properties();
// 严格主机密钥检查
sshConfig.put("StrictHostKeyChecking", "no");
sshSession.setConfig(sshConfig);
// 开启sshSession链接
sshSession.connect();
// 获取sftp通道
Channel channel = sshSession.openChannel("sftp");
// 开启
channel.connect();
sftp = (ChannelSftp) channel;
} catch (Exception e) {
logger.error("获取FTP连接实例方法异常:",e);
throw e;
}
return ftp;
}
/**
* 向FTP服务器上传文件
* @param uploadFile 本地文件全路径
* @param directory 要上传到FTP服务器的路径文件夹路径
* @param filename 要上传到FTP服务器的文件名
* @return 上传成功的文件名
* @throws Exception
*/
public String upload(String uploadFile, String directory, String filename) throws Exception {
File file = null;
String fileName = null;
InputStream in = null;
try {
createDir(directory);
file = new File(uploadFile);
in = new FileInputStream(file);
sftp.put(in, filename);
fileName = filename;
} catch (Exception e) {
logger.error("向FTP服务器上传文件方法异常:",e);
throw e;
} finally {
if (in != null) {
try {
in.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
in = null;
}
}
}
return fileName;
}
/**
* 从FTP服务器下载文件
* @param ftpdir 要从FTP下载的文件夹路径
* @param downloadFileName 要从FTP下载的文件的文件名
* @param localfilepath 要保存到本地的全路径
* @throws Exception
*/
public void download(String ftpdir, String downloadFileName, String localfilepath) throws Exception {
FileOutputStream out = null;
try {
sftp.cd(ftpdir);
File file = new File(localfilepath);
out = new FileOutputStream(file);
sftp.get(downloadFileName, out);
out.flush();
} catch (Exception e) {
logger.error("从FTP服务器下载文件方法异常:",e);
throw e;
} finally {
if (out != null) {
try {
out.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
out = null;
}
}
}
}
/**
* 从FTP服务器下载文件
* @param ftpdir 要从FTP下载的文件夹路径
* @param downloadFileName 要从FTP下载的文件的文件名
* @param localfilepath 要保存到本地的全路径
*/
public byte[] downloadSteam(String ftpdir, String downloadFileName) {
//返回的字节数组
byte[] b= null;
InputStream in = null;
try {
sftp.cd(ftpdir);
in = sftp.get(downloadFileName);
b = IOUtils.toByteArray(in);
} catch (Exception e) {
logger.error("从FTP服务器下载文件方法异常:",e);
} finally {
if (in != null) {
try {
in.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
in = null;
}
}
}
return b;
}
/**
* 列出FTP服务器指定目录下所有文件
* @param directory 需要列出的文件夹路径
* @return 对象集合
* @throws SftpException FTP异常
*/
public Vector listFiles(String directory)
throws SftpException {
return sftp.ls(directory);
}
/**
* 断开FTP连接
*/
public void disconnect() {
sftp.quit();
sftp.exit();
sftp.disconnect();
}
/**
* 判断ftp目录是否存在,如果不存在则创建目录,包括创建多级目录,并进入该目录
* @throws Exception
*/
public void createDir(String createpath) throws Exception {
try {
if (isDirExist(createpath)) {
sftp.cd(createpath);
return;
}
String pathArry[] = createpath.split("/");
StringBuffer filePath = new StringBuffer("/");
for (String path : pathArry) {
if (path.equals("")) {
continue;
}
filePath.append(path "/");
if (isDirExist(filePath.toString())) {
sftp.cd(filePath.toString());
} else {
// 建立目录
sftp.mkdir(filePath.toString());
// 进入并设置为当前目录
sftp.cd(filePath.toString());
}
}
sftp.cd(createpath);
} catch (SftpException e) {
logger.error("createDir方法异常:",e);
throw new Exception("创建路径错误:" createpath);
}
}
/**
* 判断目录是否存在
*/
public boolean isDirExist(String directory) {
boolean isDirExistFlag = false;
try {
SftpATTRS sftpATTRS = sftp.lstat(directory);
isDirExistFlag = true;
return sftpATTRS.isDir();
} catch (Exception e) {
//此处路径不存在是正常情况
logger.info("isDirExist方法判断路径--" directory "--不存在,需创建");
if (e.getMessage().toLowerCase().equals("no such file")) {
isDirExistFlag = false;
}
}
return isDirExistFlag;
}
/**
* 从FTP服务器解析错误文件返回文件内容集合---外部方法调用本方法获得文件内容,解析成数据库对象并存入数据库
* @param ftpdir
* 要从FTP下载的文件夹路径
* @param downloadFileName
* 要从FTP下载的文件的文件名
* 方案内容:通过sftp获取zip文件的字节输入流,并将该字节输入流作为构造参数构造出压缩输入流。
* 然后通过压缩输入流的getNextEntry方法,循环遍历该压缩流内部的压缩实体(即txt文件)。
* 遍历过程中将每个压缩实体转化为字节输入流,再将该字节输入流转化为字符输入流,通过IOUtils.readLines(br)读取字符输入流的内容。
* @throws Exception
*/
public Map<String,List<String>> analysisData(String ftpdir, String downloadFileName) throws Exception {
//获得缓冲输入流
BufferedInputStream bs = null;
//压缩实体
ZipEntry zipEntry = null;
//将每个压缩实体字节读入的字节数组
byte[] bytes = null;
//获得字符缓冲输入流
BufferedReader br = null;
//获取整个zip文件的输入流
InputStream inputStreamAll = null;
//将整个zip文件的输入流转为压缩输入流
ZipInputStream zipInputStream = null;
//获取单个压缩实体的输入流
InputStream inputStream=null;
//定义返回对象:key为压缩实体的文件名,value为压缩实体的内容集合
Map<String,List<String>> map = new HashMap<String,List<String>>();
try {
//进入要下载的文件路径
sftp.cd(ftpdir);
//获取整个zip文件的输入流
inputStreamAll = sftp.get(downloadFileName);
//将整个zip文件的输入流转为压缩输入流
zipInputStream = new ZipInputStream(inputStreamAll);
//当压缩文件内有文件时候进入解析
while ((zipEntry = zipInputStream.getNextEntry()) != null){
//不为文件夹时走如下逻辑
if (!zipEntry.isDirectory()) {
//当前压缩实体的文件名
String txtName = zipEntry.getName();
long size = zipEntry.getSize();
// ZipEntry的size可能为-1,表示未知
if (size == -1) {
//直接读取当前ZipEntry的流,直到为-1为止
ByteArrayOutputStream baos = new ByteArrayOutputStream();
while (true) {
int entryBytes = zipInputStream.read();
if (entryBytes == -1) break;
baos.write(entryBytes);
}
baos.close();
inputStream = new ByteArrayInputStream(baos.toByteArray());
} else { // ZipEntry的size正常
//将压缩实体通过缓冲输入流读入字节数组,然后将该字节数组转换成输入流
bytes = new byte[(int) size];
bs = new BufferedInputStream(zipInputStream);
bs.read(bytes, 0, (int) zipEntry.getSize());
inputStream = new ByteArrayInputStream(bytes);
}
//读入文件,编码格式utf-8
br = new BufferedReader(new InputStreamReader(inputStream,"UTF-8"));
//压缩实体的每行内容集合 -工具类方法
List<String> listTxt = IOUtils.readLines(br);
map.put(txtName, listTxt);
}
}
} catch (Exception e) {
logger.error("analysisData方法异常:",e);
throw e;
} finally {
if (bs != null) {
try {
bs.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
bs = null;
}
}else{
if(zipInputStream != null){
try {
zipInputStream.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
zipInputStream = null;
}
}
}
if (br != null) {
try {
br.close();
} catch (IOException io) {
io.printStackTrace();
} finally {
br = null;
}
}
disconnect();
}
return map;
}
} 来源:https://www./content-4-377501.html
|