Apache FTPClient并行读取多FTP目录报错，求可行解决方案

问题原因分析 #

1. 连接重置（SocketException）：FTP服务器检测到单连接下短时间内发起大量listFiles请求，触发了限流/主动断开机制。
2. 并行流解析异常（MalformedServerReplyException）：FTPClient并非线程安全组件，多线程共用同一个客户端实例会导致指令发送混乱，服务器响应解析失败。

解决方案 #

1. 单连接串行遍历+请求限流+重试机制 #

通过添加请求延迟、失败重试（重试前重建连接），避免触发服务器限流，同时保证线程安全：

修改getPaths方法，增加重试和延迟逻辑：

public List<String> getPaths(String path, LocalDate date) {
    int retryCount = 3;
    long delayMs = 500; // 根据服务器限流策略调整
    while (retryCount-- > 0) {
        try {
            FTPFile[] listFiles = ftp.listFiles(path);
            if (listFiles == null) {
                Thread.sleep(delayMs);
                continue;
            }
            List<String> result = Arrays.stream(listFiles)
                    .filter(f -> f.getTimestamp().getTime().toInstant()
                            .isAfter(date.atStartOfDay(ZoneId.systemDefault()).toInstant()))
                    .map(FTPFile::getName)
                    .collect(Collectors.toList());
            Thread.sleep(delayMs);
            return result;
        } catch (IOException e) {
            if (retryCount == 0) {
                e.printStackTrace();
                return Collections.emptyList();
            }
            // 重试前重建连接
            try {
                ftp.disconnect();
                ftp.connect(host, port);
                ftp.login(login, password);
                ftp.enterLocalPassiveMode(); // 确保被动模式
            } catch (IOException re) {
                re.printStackTrace();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
            return Collections.emptyList();
        }
    }
    return Collections.emptyList();
}

修改getPurchaseList为串行遍历：

public List<String> getPurchaseList(LocalDate date) {
    try (FTPClientWrapper wrapper = new FTPClientWrapper(host, port, login, password)) {
        wrapper.connect();
        wrapper.getFtpClient().enterLocalPassiveMode(); // 强制启用被动模式
        List<String> regionList = wrapper.readFile(dir, fileName);
        List<String> dirList = regionList.stream()
                .flatMap(x -> purchaseTypes.stream()
                        .map(y -> String.format("%s/%s/%s/%s", dir, x, y, folder)))
                .collect(Collectors.toList());

        List<String> allPaths = new ArrayList<>();
        for (String d : dirList) {
            List<String> paths = wrapper.getPaths(d, date);
            allPaths.addAll(paths);
        }
        return allPaths;
    } catch (IOException e) {
        e.printStackTrace();
        return Collections.singletonList(e.getMessage());
    }
}

2. 并行遍历+独立连接实例（需控制并发数） #

如果需要提升遍历速度，可使用并行流，但必须为每个线程分配独立的FTP客户端实例，同时控制并发数避免触发服务器连接数限制：

public List<String> getPurchaseListParallel(LocalDate date) {
    int maxThreads = 5; // 根据服务器最大连接数调整
    ExecutorService executor = Executors.newFixedThreadPool(maxThreads);
    try {
        // 预读取区域列表，避免并行中重复读取
        List<String> regionList;
        try (FTPClientWrapper tempWrapper = new FTPClientWrapper(host, port, login, password)) {
            tempWrapper.connect();
            regionList = tempWrapper.readFile(dir, fileName);
        }

        List<String> dirList = regionList.stream()
                .flatMap(x -> purchaseTypes.stream()
                        .map(y -> String.format("%s/%s/%s/%s", dir, x, y, folder)))
                .collect(Collectors.toList());

        List<Future<List<String>>> futures = dirList.stream()
                .map(d -> executor.submit(() -> {
                    try (FTPClientWrapper wrapper = new FTPClientWrapper(host, port, login, password)) {
                        wrapper.connect();
                        wrapper.getFtpClient().enterLocalPassiveMode();
                        Thread.sleep(300); // 添加请求间隔
                        return wrapper.getPaths(d, date);
                    } catch (Exception e) {
                        e.printStackTrace();
                        return Collections.emptyList();
                    }
                }))
                .collect(Collectors.toList());

        List<String> allPaths = new ArrayList<>();
        for (Future<List<String>> future : futures) {
            allPaths.addAll(future.get());
        }
        return allPaths;
    } catch (Exception e) {
        e.printStackTrace();
        return Collections.singletonList(e.getMessage());
    } finally {
        executor.shutdown();
    }
}

关键注意事项 #

• 被动模式必须启用：enterLocalPassiveMode()是FTP遍历目录的必要配置，避免主动模式下的连接失败。
• 调整延迟和重试次数：根据目标FTP服务器的限流策略灵活调整，平衡遍历速度和稳定性。
• 并行时控制并发数：避免因同时建立过多连接被服务器封禁。

内容的提问来源于stack exchange，提问作者user19282140