How to implement FTP batch upload files with Java multi-threading

1、构建FTP客户端

package cn.com.pingtech.common.ftp;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;

import java.io.*;
import java.net.UnknownHostException;

@Slf4j
public class  FtpConnection {

    private FTPClient ftp = new FTPClient();

    private boolean is_connected = false;

    /**
     * 构造函数
     */
    public FtpConnection() {
        is_connected = false;
        ftp.setDefaultTimeout(FtpConfig.defaultTimeoutSecond * 1000);
        ftp.setConnectTimeout(FtpConfig.connectTimeoutSecond * 1000);
        ftp.setDataTimeout(FtpConfig.dataTimeoutSecond * 1000);
        try {
            initConnect(FtpConfig.host, FtpConfig.port, FtpConfig.user, FtpConfig.password);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化连接
     *
     * @param host
     * @param port
     * @param user
     * @param password
     * @throws IOException
     */
    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();//220 连接成功
        if (!FTPReply.isPositiveCompletion(reply)) {
            disconnect();
            throw new IOException("Can't connect to server '" + host + "'");

        }
        if (!ftp.login(user, password)) {
            is_connected = false;
            disconnect();
            throw new IOException("Can't login to server '" + host + "'");
        } else {
            is_connected = true;
        }
    }

    /**
     * 上传文件
     *
     * @param path
     * @param ftpFileName
     * @param localFile
     * @throws IOException
     */
    public boolean upload(String path, String ftpFileName, File localFile) throws IOException {
        boolean is  = false;
        //检查本地文件是否存在
        if (!localFile.exists()) {
            throw new IOException("Can't upload '" + localFile.getAbsolutePath() + "'. This file doesn't exist.");
        }
        //设置工作路径
        setWorkingDirectory(path);
        //上传
        InputStream in = null;
        try {
            //被动模式
            ftp.enterLocalPassiveMode();
            in = new BufferedInputStream(new FileInputStream(localFile));
            //保存文件
            is = ftp.storeFile(ftpFileName, in);
        }catch (Exception e){
            e.printStackTrace();
        }
        finally {
            try {
                in.close();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
        return is;
    }

    /**
     * 关闭连接
     *
     * @throws IOException
     */
    public void disconnect() throws IOException {
        if (ftp.isConnected()) {
            try {
                ftp.logout();
                ftp.disconnect();
                is_connected = false;
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    /**
     * 设置工作路径
     *
     * @param dir
     * @return
     */
    private boolean setWorkingDirectory(String dir) {
        if (!is_connected) {
            return false;
        }
        //如果目录不存在创建目录
        try {
            if (createDirecroty(dir)) {
                return ftp.changeWorkingDirectory(dir);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;

    }

    /**
     * 是否连接
     *
     * @return
     */
    public boolean isConnected() {
        return is_connected;
    }

    /**
     * 创建目录
     *
     * @param remote
     * @return
     * @throws IOException
     */
    private boolean createDirecroty(String remote) throws IOException {
        boolean success = true;
        String directory = remote.substring(0, remote.lastIndexOf("/") + 1);
        // 如果远程目录不存在，则递归创建远程服务器目录
        if (!directory.equalsIgnoreCase("/") && !ftp.changeWorkingDirectory(new String(directory))) {
            int start = 0;
            int end = 0;
            if (directory.startsWith("/")) {
                start = 1;
            } else {
                start = 0;
            }
            end = directory.indexOf("/", start);
            while (true) {
                String subDirectory = new String(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);
                // 检查所有目录是否创建完毕
                if (end <= start) {
                    break;
                }
            }
        }
        return success;
    }

}

2、FTP连接工厂

package cn.com.pingtech.common.ftp;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;


/**
 * 连接工厂

 */
@Slf4j
public class FtpFactory {

    //有界队列
    private static final ArrayBlockingQueue<FtpConnection> arrayBlockingQueue = new ArrayBlockingQueue<>(FtpConfig.ftpConnectionSize);


    protected FtpFactory(){
        log.info("init ftpConnectionSize "+FtpConfig.ftpConnectionSize);
        for(int i = 0; i< FtpConfig.ftpConnectionSize; i++){
            //表示如果可能的话，将 e 加到 BlockingQueue 里，即如果 BlockingQueue 可以容纳，则返回 true，否则返回 false
            arrayBlockingQueue.offer(new FtpConnection());
        }
    }

    /**
     * 获取连接
     *
     * @return
     */

    public FtpConnection getFtp() {
        FtpConnection poll = null;
        try {
            //取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 Blocking 有新的对象被加入为止
            poll = arrayBlockingQueue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return poll;
    }

    /**
     * 释放连接
     * @param ftp
     * @return
     */
    public boolean relase(FtpConnection ftp){
        return arrayBlockingQueue.offer(ftp);
    }

    /**
     * 删除连接
     *
     * @param ftp
     */

    public void remove(FtpConnection ftp) {
        arrayBlockingQueue.remove(ftp);
    }

    /**
     * 关闭连接
     */
    public void close() {
        for (FtpConnection connection : arrayBlockingQueue) {
            try {
                connection.disconnect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


}

3、FTP配置

package cn.com.pingtech.common.ftp;

/**
 * ftp 配置类
 */

public class FtpConfig {

    public static int defaultTimeoutSecond = 10;
    public static int connectTimeoutSecond = 10;
    public static int dataTimeoutSecond = 10;
    public static String host = "127.0.0.1";
    public static int port =9999;
    public static String user = "Administrator";
    public static String password ="Yp886611";
    public static int threadPoolSize = 1;
    public static int ftpConnectionSize = 1;
    
}

4、构建多线程FTP上传任务

package cn.com.pingtech.common.ftp;

import java.io.File;
import java.io.IOException;
import java.util.concurrent.Callable;


/**
 * 上传任务
 */
public class UploadTask implements Callable{
    private File file;

    private FtpConnection ftp;

    private String path;

    private String fileName;

    private FtpFactory factory;

    public UploadTask(FtpFactory factory,FtpConnection ftp, File file, String path, String fileName){

        this.factory = factory;

        this.ftp = ftp;

        this.file = file;

        this.path = path;

        this.fileName = fileName;

    }

    @Override
    public UploadResult call() throws Exception {
        UploadResult result = null;
        try {
            if (ftp == null) {
                result = new UploadResult(file.getAbsolutePath(), false);
                return result;
            }
            //如果连接未开启 重新获取连接
            if (!ftp.isConnected()) {
                factory.remove(ftp);
                ftp = new FtpConnection();
            }

            //开始上传
            result = new UploadResult(file.getName(), ftp.upload(path, fileName, file));
        } catch (IOException ex) {
            result = new UploadResult(file.getName(), false);
            ex.printStackTrace();
        } finally {
            factory.relase(ftp);//释放连接
        }
        return result;

    }
}

package cn.com.pingtech.common.ftp;
/**
 * 上传结果
 */
public class UploadResult {
    private String fileName; //文件名称
    private boolean result; //是否上传成功

    public UploadResult(String fileName, boolean result) {
        this.fileName = fileName;
        this.result = result;
    }

    public String getFileName() {
        return fileName;

    }

    public void setFileName(String fileName) {
        this.fileName = fileName;
    }

    public boolean isResult() {
        return result;
    }

    public void setResult(boolean result) {
        this.result = result;
    }

    public String toString() {
        return "[fileName=" + fileName + " , result=" + result + "]";
    }
}

注意：实现Callable接口的任务线程能返回执行结果
Callable接口支持返回执行结果，此时需要调用FutureTask.get()方法实现，此方法会阻塞线程直到获取“将来”的结果，当不调用此方法时，主线程不会阻塞

5、FTP上传工具类

package cn.com.pingtech.common.ftp;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


/**
 * ftp上传工具包
 */

public class FtpUtil {

    /**
     * 上传文件
     *
     * @param ftpPath
     * @param listFiles
     * @return
     */

    public static synchronized List upload(String ftpPath, File[] listFiles) {
        //构建线程池
        ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(FtpConfig.threadPoolSize);
        List<Future> results = new ArrayList<>();
        //创建n个ftp链接
        FtpFactory factory = new FtpFactory();
        for (File file : listFiles) {
            FtpConnection ftp = factory.getFtp();//获取ftp con
            UploadTask upload = new UploadTask(factory,ftp, file, ftpPath, file.getName());
            Future submit = newFixedThreadPool.submit(upload);
            results.add(submit);
        }

        List listResults = new ArrayList<>();
        for (Future result : results) {
            try {
                //获取线程结果
                UploadResult uploadResult = (UploadResult)result.get(30, TimeUnit.MINUTES);
                listResults.add(uploadResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        factory.close();
        newFixedThreadPool.shutdown();
        return listResults;
    }

}

6、测试上传

package cn.com.pingtech.common.ftp


class Client {
    public static void main(String[] args) throws IOException {
        String loalPath = "C:\\Users\\Administrator\\Desktop\\test\\0";
        String ftpPath = "/data/jcz/";
        File parentFile = new File(loalPath);
        List <UploadResult> list = FtpUtil.upload(ftpPath,parentFile.listFiles());
        for(UploadResult vo:list){
            System.out.println(vo);
        }
        
    }
}

注意：FTP协议里面，规定文件名编码为iso-8859-1，所以目录名或文件名需要转码

The above is the detailed content of How to implement FTP batch upload files with Java multi-threading. For more information, please follow other related articles on the PHP Chinese website!