某项目由于网络关系，传输文件通过ftp方式，即第三方实时把碎片文件放到ftp，我方需要把ftp文件下载下来进行业务处理。每个文件在100KB左右，每日文件数量在300万左右。前同事直接用在网上找的ftp demo，每次上传/下载文件创建新的连接，且单线程处理文件导致处理性能在1个文件每秒左右。当我们发现时，堆积的文件已经快撑爆了ftp服务器。
所以需要对现有代码进行优化。将现有的单线程改为多线程，每处理一个文件就启动一个线程来处理。创建ftp连接池，为每个线程提供单独的ftp连接。当线程下载/上传完成一个文件后，立即将连接还给Ftp连接池，提供连接池复用效率。
优化Ftp服务器配置
如果采用多线程消费ftp上文件，并且采用ftp连接池维护ftp连接，需要确认ftp服务的配置是否支持多个连接。项目采用vsftpd，所以需要查看/etc/vsftpd/vsftpd.conf配置文件。
- pasv_max_port=0 设置在PASV工作方式下，数据连接可以使用的端口范围的上界。默认值为0，表示任意端口。
- pasv_mim_port=0 设置在PASV工作方式下，数据连接可以使用的端口范围的下界。默认值为0，表示任意端口。
- max_clients=0 设置vsftpd允许的最大连接数，默认为0，表示不受限制。若设置为150时，则同时允许有150个连接，超出的将拒绝建立连接。只有在以standalone模式运行时才有效。
- max_per_ip=0 设置每个IP地址允许与FTP服务器同时建立连接的数目。默认为0，不受限制。通常可对此配置进行设置，防止同一个用户建立太多的连接。只有在以standalone模式运行时才有效。
配置完成后重启ftp服务。如果ftp连接失败，可以检查是否是防火墙/安全策略限制了端口放行。
连接池
引入Maven
<!-- https://mvnrepository.com/artifact/commons-net/commons-net -->
<dependency>
<groupId>commons-net</groupId>
<artifactId>commons-net</artifactId>
<version>3.10.0</version>
</dependency>
Ftp连接配置
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
@Configuration
@Data
public class FtpProperties {
@Value("${ftp.hostname}")
private String host;
@Value("${ftp.username}")
private String username;
@Value("${ftp.password}")
private String password;
@Value("${ftp.port}")
private int port;
@Value("${ftp.timeout.default.seconds:1200}")
private Integer defaultTimeoutSecond = 1200;
@Value("${ftp.timeout.connect.seconds:1800}")
private Integer connectTimeoutSecond = 1800;
@Value("${ftp.timeout.data.seconds:2400}")
private Integer dataTimeoutSecond = 2400;
@Value("${ftp.charSet:UTF-8}")
private String charSet = "UTF-8";
@Value("${ftp.factory.thread.times:2}")
private Integer threadTimes = 2;
}
Ftp连接对象
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import java.io.IOException;
import java.io.OutputStream;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* ftp连接
*/
@Slf4j
public class FtpConnection {
private final FTPClient ftp = new FTPClient();
private final AtomicBoolean isConnected = new AtomicBoolean(false);
/**
* 构造函数
*/
public FtpConnection(FtpProperties ftpProperties) {
ftp.setDefaultTimeout(ftpProperties.getDefaultTimeoutSecond() * 1000);
ftp.setConnectTimeout(ftpProperties.getConnectTimeoutSecond() * 1000);
ftp.setDataTimeout(ftpProperties.getDataTimeoutSecond() * 1000);
ftp.setCharset(Charset.forName(ftpProperties.getCharSet()));
ftp.setControlEncoding(ftpProperties.getCharSet());
//被动模式
ftp.enterLocalPassiveMode();
try {
initConnect(ftpProperties.getHost(), ftpProperties.getPort(),
ftpProperties.getUsername(), ftpProperties.getPassword());
} catch (IOException e) {
isConnected.set(false);
log.error("init ftp client error", e);
}
}
/**
* 初始化连接
*/
private void initConnect(String host, int port, String user, String password) throws IOException {
try {
ftp.connect(host, port);
} catch (UnknownHostException ex) {
throw new IOException("Can't find FTP server '" + host + "'");
}
int reply = ftp.getReplyCode();
if (!FTPReply.isPositiveCompletion(reply)) {
disconnect();
throw new IOException("Can't connect to server '" + host + "'");
}
if (!ftp.login(user, password)) {
isConnected.set(false);
disconnect();
throw new IOException("Can't login to server '" + host + "'");
} else {
isConnected.set(true);
}
}
public List<String> fileNames(String path) {
try {
//获取文件列表使用这个接口最好，命令行效率更高
String[] strings = ftp.listNames(path);
return strings == null ? new ArrayList<>() : Stream.of(strings).collect(Collectors.toList());
} catch (IOException e) {
log.error("ftp listNames error.path = {}", path);
return new ArrayList<>();
}
}
public String download(String path, String ftpFileName, String localPath, boolean deleteSuccessFile, boolean deleteErrorFile) {
String fileName = localPath + ftpFileName.substring(path.length());
boolean deleted = false;
try (OutputStream os = Files.newOutputStream(Paths.get(fileName))) {
//保存文件
boolean success = ftp.retrieveFile(ftpFileName, os);
if (success) {
//是否需要删除源文件
if (deleteSuccessFile) {
deleted = delete(ftpFileName);
}
return fileName;
}
} catch (Exception e) {
log.error("ftp download error", e);
if (deleteErrorFile) {
//对于下载失败的文件直接删除
deleted = delete(ftpFileName);
}
} finally {
if (deleteSuccessFile || deleteErrorFile) {
log.debug("ftp download then delete file {}", deleted ? "success" : "failed");
}
}
return null;
}
public boolean delete(String ftpFileName) {
try {
ftp.deleteFile(ftpFileName);
return true;
} catch (IOException e) {
log.error("delete ftp file error", e);
}
return false;
}
/**
* 关闭连接
*/
public void disconnect() {
if (ftp.isConnected()) {
try {
ftp.logout();
ftp.disconnect();
isConnected.set(false);
} catch (IOException e) {
log.error("ftp disconnect error", e);
}
}
}
/**
* 设置工作路径
*/
private boolean setWorkingDirectory(String dir) {
if (!isConnected.get()) {
return false;
}
//如果目录不存在创建目录
try {
if (createDirectory(dir)) {
return ftp.changeWorkingDirectory(dir);
}
} catch (IOException e) {
log.error("set working directory error", e);
}
return false;
}
/**
* 是否连接
*/
public boolean isConnected() {
return isConnected.get();
}
/**
* 创建目录
*/
private boolean createDirectory(String remote) throws IOException {
boolean success = true;
String directory = remote.substring(0, remote.lastIndexOf("/") + 1);
// 如果远程目录不存在，则递归创建远程服务器目录
if (!directory.equalsIgnoreCase("/") && !ftp.changeWorkingDirectory(directory)) {
int start = 0;
int end;
if (directory.startsWith("/")) {
start = 1;
}
end = directory.indexOf("/", start);
do {
String subDirectory = remote.substring(start, end);
if (!ftp.changeWorkingDirectory(subDirectory)) {
if (ftp.makeDirectory(subDirectory)) {
ftp.changeWorkingDirectory(subDirectory);
} else {
log.error("mack directory error :/" + subDirectory);
return false;
}
}
start = end + 1;
end = directory.indexOf("/", start);
// 检查所有目录是否创建完毕
} while (end > start);
}
return success;
}
}
Ftp工厂
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;
/**
* ftp连接工厂
*/
@Slf4j
@NoArgsConstructor
public class FtpFactory {
/**
* 存放ftp连接有界队列
*/
private ArrayBlockingQueue<FtpConnection> arrayBlockingQueue;
private FtpProperties ftpProperties;
/**
* 最大连接数在初始化的时候固定，除非删除，否则不能新增连接
*/
private int maxSize;
/**
* 工厂含有的连接数(包括已经取出的)
*/
private int ftpSize = 0;
private final ReentrantLock lock = new ReentrantLock(false);
public FtpFactory(FtpProperties ftpProperties) {
this.ftpProperties = ftpProperties;
//最大数量为处理器线程数*配置的倍数
this.maxSize = Runtime.getRuntime().availableProcessors() * ftpProperties.getThreadTimes();
this.arrayBlockingQueue = new ArrayBlockingQueue<>(maxSize);
//初始化一个客户端，节约资源。后续需要更多再追加。
this.fill(1);
}
/**
* 扩容工厂的连接池
*
* @param size 扩容连接数量至
*/
public void fill(int size) {
if (size <= 0) {
return;
}
final ReentrantLock lock = this.lock;
lock.lock();
try {
for (int i = 0; i < size; i++) {
if (ftpSize >= size || ftpSize >= maxSize) {
break;
}
//表示如果可能的话，将 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false
FtpConnection connection = new FtpConnection(ftpProperties);
boolean offer = this.safeOffer(connection);
if (!offer) {
break;
} else {
ftpSize++;
}
}
log.info("Fill ftpConnection end, size is {}.", ftpSize);
} finally {
lock.unlock();
}
}
/**
* 将ftp连接放入队列，如果超过队列大小将会销毁连接
*
* @param connection
* @return 是否加入成功
*/
private boolean safeOffer(FtpConnection connection) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
boolean offer = arrayBlockingQueue.offer(connection);
if (!offer) {
log.debug("offer ftpConnection failed");
connection.disconnect();
}
return offer;
} finally {
lock.unlock();
}
}
/**
* 获取连接(阻塞)
*/
public FtpConnection getFtp() {
FtpConnection poll;
try {
//取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 Blocking 有新的对象被加入为止
poll = arrayBlockingQueue.take();
} catch (InterruptedException e) {
log.error("getFtpConnection error", e);
return null;
}
return poll;
}
/**
* 释放连接
*
* @param ftp
*/
public void safeRelease(FtpConnection ftp) {
this.safeOffer(ftp);
}
/**
* 关闭连接
*/
public void close() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
for (FtpConnection connection : arrayBlockingQueue) {
connection.disconnect();
}
arrayBlockingQueue.clear();
ftpSize = 0;
} finally {
lock.unlock();
}
}
}
工厂使用
使用方式
- 获取连接 FtpConnection connection = ftpFactory.getFtp();
- 归还连接 ftpFactory.safeRelease(connection);
- 扩容连接池 ftpFactory.fill(50); 50代表想要50个连接，注意有最大限制。最大限制为FtpFactory中的maxSize，可以通过ftp.factory.thread.times配置为系统线程数的倍数。
代码示例
public void yourJob(FtpFactory ftpFactory, String ftpFilePath) {
try {
//拿到一个连接
FtpConnection connection = ftpFactory.getFtp();
List<String> names = connection.fileNames(ftpFilePath);
//归还连接
ftpFactory.safeRelease(connection);
if (!ObjectUtils.isEmpty(names)) {
ftpFactory.fill(names.size());
CountDownLatch countDownLatch = new CountDownLatch(names.size());
Logger.info("本次文件路径 {}, 总数 {}, 开始处理", ftpFilePath, names.size());
long start = System.currentTimeMillis();
for (String fileName : names) {
ftpThreadPool.execute(
() -> {
//这里传入ftpFactory是为了从工厂拿到连接和归还连接
yourService.downloadAndAnalysis(ftpFactory, ftpFilePath, fileName);
countDownLatch.countDown();
});
}
countDownLatch.await();
double seconds = (System.currentTimeMillis() - start) / 1000.0;
Logger.info("完成文件路径 {}, 总数 {}, 耗时 {} 秒, 速度 {} 条/s",
ftpFilePath, names.size(), seconds, names.size() / seconds);
} else {
Logger.error("ftp文件为空ftpFilePath：{}", ftpFilePath);
}
} catch (Exception e) {
Logger.error("同步图片数据异常", e);
}
}