Apache FTPClient并行操作的陷阱与解决方案：多线程连接管理

// 假设这是在并行任务中执行的代码
public List<String> getPathsConcurrently(String host, int port, String login, String password, String path, LocalDate date) {
    FTPClient ftp = new FTPClient();
    try {
        ftp.connect(host, port);
        ftp.login(login, password);
        ftp.enterLocalPassiveMode(); // 推荐使用被动模式

        FTPFile[] listFiles = ftp.listFiles(path);
        return Arrays.stream(listFiles)
                .filter(f -> f.getTimestamp().getTime().toInstant().isAfter(date.atStartOfDay(ZoneId.systemDefault()).toInstant()))
                .map(FTPFile::getName)
                .collect(Collectors.toList());
    } catch (IOException e) {
        e.printStackTrace();
        return Collections.emptyList();
    } finally {
        try {
            if (ftp.isConnected()) {
                ftp.logout();
                ftp.disconnect();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

// 在 getPurchaseList 中调用时，每个并行任务都创建新连接
public List<String> getPurchaseList(LocalDate date, String host, int port, String login, String password) {
    // ... dirList 的生成逻辑不变 ...

    return dirList.parallelStream() // 使用 parallelStream 实现并行
            .flatMap(d -> getPathsConcurrently(host, port, login, password, d, date).stream())
            .collect(Collectors.toList());
}

import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.io.IOException;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

// 1. FTPClientFactory 实现
class FtpClientPooledObjectFactory implements PooledObjectFactory<FTPClient> {
    private final String host;
    private final int port;
    private final String user;
    private final String password;

    public FtpClientPooledObjectFactory(String host, int port, String user, String password) {
        this.host = host;
        this.port = port;
        this.user = user;
        this.password = password;
    }

    @Override
    public PooledObject<FTPClient> makeObject() throws Exception {
        FTPClient ftp = new FTPClient();
        ftp.connect(host, port);
        if (!ftp.login(user, password)) {
            throw new IOException("Failed to login to FTP server.");
        }
        ftp.enterLocalPassiveMode(); // 推荐使用被动模式
        ftp.setFileType(FTPClient.BINARY_FILE_TYPE); // 根据需要设置文件类型
        return new DefaultPooledObject<>(ftp);
    }

    @Override
    public void destroyObject(PooledObject<FTPClient> p) throws Exception {
        FTPClient ftp = p.getObject();
        if (ftp.isConnected()) {
            ftp.logout();
            ftp.disconnect();
        }
    }

    @Override
    public boolean validateObject(PooledObject<FTPClient> p) {
        FTPClient ftp = p.getObject();
        try {
            return ftp.sendNoOp(); // 发送一个NOOP命令来验证连接是否活跃
        } catch (IOException e) {
            return false;
        }
    }

    @Override
    public void activateObject(PooledObject<FTPClient> p) throws Exception {
        // 可选：在从池中借出时执行一些操作
    }

    @Override
    public void passivateObject(PooledObject<FTPClient> p) throws Exception {
        // 可选：在归还到池中时执行一些操作，例如重置工作目录
    }
}

// 2. FTPClient连接池管理器
class FtpClientPoolManager implements AutoCloseable {
    private final GenericObjectPool<FTPClient> pool;

    public FtpClientPoolManager(String host, int port, String user, String password, int maxTotal) {
        FtpClientPooledObjectFactory factory = new FtpClientPooledObjectFactory(host, port, user, password);
        GenericObjectPoolConfig<FTPClient> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(maxTotal); // 最大连接数
        config.setBlockWhenExhausted(true); // 当池耗尽时是否阻塞
        config.setMaxWaitMillis(5000); // 阻塞等待时间
        config.setTestOnBorrow(true); // 借出时验证连接
        config.setTestOnReturn(true); // 归还时验证连接

        this.pool = new GenericObjectPool<>(factory, config);
    }

    public FTPClient borrowClient() throws Exception {
        return pool.borrowObject();
    }

    public void returnClient(FTPClient client) {
        if (client != null) {
            pool.returnObject(client);
        }
    }

    @Override
    public void close() {
        pool.close();
    }
}

// 3. 修改后的 getPaths 方法，从连接池获取客户端
public class FtpParallelProcessor {

    private final FtpClientPoolManager poolManager;

    public FtpParallelProcessor(String host, int port, String user, String password, int maxConnections) {
        this.poolManager = new FtpClientPoolManager(host, port, user, password, maxConnections);
    }

    public List<String> getPathsFromPool(String path, LocalDate date) {
        FTPClient ftp = null;
        try {
            ftp = poolManager.borrowClient(); // 从连接池获取客户端
            FTPFile[] listFiles = ftp.listFiles(path);
            return Arrays.stream(listFiles)
                    .filter(f -> f.getTimestamp().getTime().toInstant().isAfter(date.atStartOfDay(ZoneId.systemDefault()).toInstant()))
                    .map(FTPFile::getName)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            e.printStackTrace();
            return Collections.emptyList();
        } finally {
            poolManager.returnClient(ftp); // 归还客户端到连接池
        }
    }

    // 修改后的 getPurchaseList 方法
    public List<String> getPurchaseList(LocalDate date, List<String> dirList) {
        return dirList.parallelStream()
                .flatMap(d -> getPathsFromPool(d, date).stream())
                .collect(Collectors.toList());
    }

    // 关闭连接池
    public void closePool() throws Exception {
        poolManager.close();
    }
}