这篇文章主要是想记录一下通过java实现一个通用的连接SFTP上传文件的功能。
我们在项目中经常会遇到本系统和其他系统之间的交互需要靠文件来实现，我们自己的系统将内容写到文件上传到sftp，其他系统从sftp下载结息数据。
比如，我们自己的系统上传a.txt文件到sftp,但是有个问题，如果文件特别大，文件生成了但是怎么确定文件内容是输出完整的，这个时候一般我们会约定在生成一个文件，用来标识a.txt的内容输出完成。比如当a.txt内容写完之后生成a.txt.ok文件用来标识内容的完整。
不同的文件本质上是存储着不同对象的字段，所以不同的实现只需要继承父类就可以了。在父类中写了主要的通用方法，子类只需要实现各自的特殊点就可以了。
public class UploadFileToSftpBase {
private ThreadLocal<ChannelSftp> channelSftpThreadLocal = new ThreadLocal<>();
// 这个方法主要是将不同的对象集合内容，先写到一个本地路径下的文件中
public String genStatementFile(List<UploadFileToSftpBase> details, Date statementDate, String fileName, String delimiter, String lineEnd,Boolean includeHeader) {
// 将实体类对象转成生成文件需要的字段
List<UploadFileToSftpBase> dataList = convert2CsvDtoList(details);
//显示顺序
String[] columnMapping = getColumnMapping();
//表头
String[] header = getHeader();
String localTempPath = "/usr/local/temp/时间戳";
FileUtil.write(dataList,
UploadFileToSftpBase.class,
localTempPath,
columnMapping,
header,
includeHeader,
delimiter,
lineEnd);
return localTempPath;
}
public List<UploadFileToSftpBase> convert2CsvDtoList(List<UploadFileToSftpBase> dataList) {
return dataList.stream().map(d -> convert2CsvDto(d)).collect(Collectors.toList());
}
// 如果集合中的实体和最终需要的字段有差别，可以实现方法进行转换
public UploadFileToSftpBase convert2CsvDto(UploadFileToSftpBasedetail) {
return detail;
}
/**
* 列对应关系
*/
public String[] getColumnMapping() {
String[] columnMapping = {
"name",
"age"
};
return columnMapping;
}
/**
* 表头
*/
public String[] getHeader() {
return getColumnMapping();
}
// 暴露的上传方法
public void uploadToSftp(String uploadFilePath, String filePath) {
String okFileLocalName = createOKFile(uploadFilePath, filePath);
String okFileRemoteName = Paths.get(new File(uploadFilePath).getParent(),
new File(okFileLocalName).getName()).toString();
sftp = getSftp(getSftpConfig)
upload(filePath, uploadFilePath);
upload(okFileLocalName, okFileRemoteName);
}
new File(filePath).delete();
new File(okFileLocalName).delete();
}
// 真正使用时 这些内容作为配置参数放在配置文件中
public SftpConfig getSftpConfig() {
SftpConfig sftpConfig = new SftpConfig();
// 这个ip为B机器所在ip
String host = "100.100.932.267";
// 这个端口为B机器nginx所监听的端口
int port = 66;
String user = "sftp_user";
String password = "sftp_passward";
sftpConfig.setHost(host);
sftpConfig.setPort(port);
sftpConfig.setUsername(user);
sftpConfig.setPassword(password);
return sftpConfig;
}
// 获取sftp连接
@PostConstruct
public ChannelSftp getSftp(SftpConfig config){
if(channelSftpThreadLocal.get()!=null){
return channelSftpThreadLocal.get();
}else{
JSch jsch = new JSch();
Session session = jsch.getSession(config.getUser(), config.getHost(), config.getPort());
session.setPassword(config.getPassword());
session.setConfig("StrictHostKeyChecking", "no");
session.connect();
ChannelSftp channelSftp = (ChannelSftp) session.openChannel("sftp");
channelSftp.connect();
channelSftpThreadLocal.set(channelSftp );
return channelSftp;
}
}
// 空方法
@Override
public void process(JSONObject paramDto) throws Exception {
}
// 上传本地本见到sftp
public void upload(String srcFile, String destFile) {
try {
Path destPath = Paths.get(destFile);
Path parent = destPath.getParent();
String fileName = destPath.getFileName().toString();
cdRoot();
if (parent != null) {
// cd parent folder
tryMkdir(channelSftpThreadLocal.get(), parent.toString(), 1);
}
File file = new File(srcFile);
try (FileInputStream steam = new FileInputStream(file)) {
// 不使用 destFile 目的是为了兼任windows+linux;
String partFileName = fileName + ".processing";
channelSftpThreadLocal.get().put(steam, partFileName);
// 使用重命名确保文件完整
channelSftpThreadLocal.get().rename(partFileName, fileName);
}
} catch (Exception e) {
throw new ProcessException("sftp 上传文件失败", e);
}
}
// 进到对象路径
@SneakyThrows
public void cdRoot() {
String rootDir = getSftpRootDir();
channelSftpThreadLocal.get().cd(rootDir);
}
// 如果路径不存在创建
private void tryMkdir(ChannelSftp channel, String desFolder, int level) {
if (StringUtils.isEmpty(desFolder)) {
return;
}
if (level > 10) {
return;
}
Path path = Paths.get(desFolder);
if (level > path.getNameCount()) {
return;
}
String currentDir = path.getName(level - 1).toString();
// 如果文件夹不存在, 先创建文件夹再进入。
try {
channel.cd(currentDir);
} catch (Exception e) {
try {
channel.mkdir(currentDir);
channel.cd(currentDir);
} catch (Exception ex) {
log.info("mkdir error:{}", ex.getMessage());
}
}
tryMkdir(channel, desFolder, level + 1);
}
将实体类对象写到本地文件中，定义了列之间、行之间的分隔符
public class FileUtil{
/**
* columnMapping 字段映射
* header 文件表头 includeHeader 是否包含表头
* delimiter 每一列的分隔符 lineEnd 每一行的分隔符
*/
public static <T> void write(List<T> data, Class<T> classOfT, String filePath, String[] columnMapping, String[] header,
Boolean includeHeader, String delimiter,String lineEnd) {
char separator = delimiter.charAt(0);
try (Writer writer = new FileWriter(filePath);
CSVWriter csvWriter = new CSVWriter(writer, separator,
CSVWriter.NO_QUOTE_CHARACTER, CSVWriter.DEFAULT_ESCAPE_CHARACTER, CSVWriter.RFC4180_LINE_END);) {
// //加上BOM标识
// writer.write(new String(new byte[]{(byte) 0xEF, (byte) 0xBB, (byte) 0xBF}));
ColumnPositionMappingStrategy<T> mapper =
new ColumnPositionMappingStrategy<>();
mapper.setType(classOfT);
mapper.setColumnMapping(columnMapping);
if (includeHeader) {
csvWriter.writeNext(header);
}
StatefulBeanToCsv<T> beanToCsv = new StatefulBeanToCsvBuilder<T>(writer)
.withMappingStrategy(mapper)
.withQuotechar(CSVWriter.NO_QUOTE_CHARACTER)
.withSeparator(separator)
.withLineEnd(lineEnd)
.withEscapechar('\\').build();
beanToCsv.write(data);
} catch (Exception e) {
LOGGER.error(ExceptionUtils.getStackTrace(e));
throw new FailNoticeException("文件解析失败.", e);
}
}
}
不同的文件使用不同的实体类，但是都需要继承父类
public class User extends UploadFileToSftpBase implements Serializable {
static final long serialVersionUID = 42L;
private String name;
private int age;
}
@Slf4j
@Service
@JobHandler("TestJob ")
public class TestJob extends UploadFileToSftpBase{
@Override
public void process() throws Exception {
// 1.抓数据
User object = new User ();
object.setName("王麻子");
object.setAge(18);
String dateStr = "20240112";
User object2 = new User ();
object2.setName("赵倩");
object2.setAge(20);
// 2.生成CSV数据
String fileName = String.format("XXX_VVV_INFO_%s.txt", dateStr);
String localPath = genStatementFile(CollectionUtil.toList(object, object2), new Date(), fileName, "|", "@@", false);
String uploadPath = String.format("/BB/%s/DD/%s", dateStr, fileName);
// 3.上传
uploadToSftp(uploadPath, localPath);
}
@Override
public UploadFileToSftpBase convert2CsvDto(UploadFileToSftpBase detail) {
return detail;
}
@Override
public String[] getColumnMapping() {
String[] columnMapping = {
"name",
"age"
};
return columnMapping;
}