利用ftp多线程上传文件

Posted 2022-08-31 褚金辉

使用apache commons-net-3.3

首先我们需要有一个ftp服务器。

直接上代码

package com.ourpalm.resupgrade.util.ftp;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.UnknownHostException;

import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;

import com.ourpalm.resupgrade.config.FtpConfig;

/**
 * FTP 连接
 * @author chuer
 * @date 2015年1月7日 下午2:19:48
 */
public class FtpConnection 
    public static final String ANONYMOUS_LOGIN = "anonymous";
    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();
        if (!FTPReply.isPositiveCompletion(reply)) 
            disconnect();
            throw new IOException("Can't connect to server '" + host + "'");
        

        if (user == "") 
            user = ANONYMOUS_LOGIN;
        

        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 void upload(String path,String ftpFileName, File localFile) throws IOException 
        //检查本地文件是否存在
        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));
            //保存文件
            if (!ftp.storeFile(ftpFileName, in)) 
                throw new IOException("Can't upload file '" + ftpFileName + "' to FTP server. Check FTP permissions and path.");
            
         finally 
            try 
                in.close();
             catch (IOException ex) 
            
        
    

    /**
     * 关闭连接
     * @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 
						System.out.println("mack directory error :/"+subDirectory);
						return false;
					
				
				start = end + 1;
				end = directory.indexOf("/", start);
				// 检查所有目录是否创建完毕
				if (end <= start) 
					break;
				
			
		
		return success;
	
    

package com.ourpalm.resupgrade.util.ftp;
import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;

import com.ourpalm.resupgrade.config.FtpConfig;

/**
 * 连接工厂
 * @author chuer
 * @date 2015年1月7日 下午2:32:22
 */
public class FtpFactory 
	private final ArrayBlockingQueue<FtpConnection> arrayBlockingQueue = new ArrayBlockingQueue<>(FtpConfig.ftpConnectionSize);
	
	protected FtpFactory()
		System.out.println("init  FtpFactory");
		for(int i=0;i< FtpConfig.ftpConnectionSize; i++)
			arrayBlockingQueue.offer(new FtpConnection());
		
	
	
	/**
	 * 获取连接
	 * @return
	 */
	public  FtpConnection getFtp()
		FtpConnection poll = null;
		try 
			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();
			
		
		
	
	

 

package com.ourpalm.resupgrade.util.ftp;

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

import com.ourpalm.resupgrade.util.log.LoggerUtils;

/**
 * 上传任务
 * @author chuer
 * @date 2015年1月7日 下午2:30:46
 */
public class UploadTask implements Callable<UploadResult>
	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();
//				factory.relase(ftp);
			
			
			//开始上传
			LoggerUtils.upload.info(file.getName()+" is uploading ...");
			
			FtpResult.resultList.add(file.getName()+" is uploading ...");
			ftp.upload(path, fileName, file);
			result = new UploadResult(file.getName(),true);
		catch(IOException ex)
			result = new UploadResult(file.getName(),false);
			ex.printStackTrace();
		finally
			factory.relase(ftp);//释放连接
		
		
		FtpResult.resultList.add(result.toString());
		LoggerUtils.upload.info(result.toString());
		return result;
	

	

 

package com.ourpalm.resupgrade.util.ftp;

/**
 * 上传结果
 * @author chuer
 * @date 2015年1月7日 下午2:31:14
 */
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+"]";
	
	
	

 

package com.ourpalm.resupgrade.util.ftp;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 实时上传结果
 * @author chuer
 * @date 2015年1月7日 下午4:57:24
 */
public class FtpResult 
	public static List<String> resultList = new CopyOnWriteArrayList<>();
	

 

package com.ourpalm.resupgrade.util.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;

import com.ourpalm.resupgrade.config.FtpConfig;

/**
 * ftp上传工具包
 * @author chuer
 * @date 2015年1月7日 下午2:31:39
 */
public class FtpUtil 

	/**
	 * 上传文件
	 * @param ftpPath
	 * @param listFiles
	 * @return
	 */
	public static synchronized List<UploadResult> upload(String ftpPath,File [] listFiles) 
		ExecutorService newFixedThreadPool = Executors .newFixedThreadPool(FtpConfig.threadPoolSize);
		List<Future<UploadResult>> results = new ArrayList<>();
		
		FtpFactory factory = new FtpFactory();
		for (File file : listFiles) 
			FtpConnection ftp = factory.getFtp();
			
			UploadTask upload = new UploadTask(factory,ftp, file, ftpPath, file.getName());
			Future<UploadResult> submit = newFixedThreadPool.submit(upload);
			results.add(submit);
		

		
		List<UploadResult> listResults = new ArrayList<>();
		for (Future<UploadResult> result : results) 
			try 
				UploadResult uploadResult = result.get(30, TimeUnit.MINUTES);
				listResults.add(uploadResult);
			 catch (Exception e) 
				e.printStackTrace();
			
		
		factory.close();
		newFixedThreadPool.shutdown();
		return listResults;
	

 

package com.ourpalm.resupgrade.config;

import org.w3c.dom.Document;
import org.w3c.dom.Element;

import com.ourpalm.resupgrade.util.io.XmlUtils;
/**
 * ftp 配置类
 * @author chuer
 * @date 2015年1月7日 下午4:36:50
 */
public class FtpConfig 
	
	public static  int defaultTimeoutSecond;
	public static  int connectTimeoutSecond;
	public static  int dataTimeoutSecond;
	public static  String host;
	public static  int port;
	public static  String user;
	public static  String password;
	
    
	public static int threadPoolSize;
	public static int ftpConnectionSize;
	
	public static  String rootPath;
    
    /**
     * @param path
     */
    public static void load(String path) 
		path += "ftpConfig.xml";
		try
			Document doc = XmlUtils.load(path);
			Element root = doc.getDocumentElement();
			
			defaultTimeoutSecond = Integer.parseInt(XmlUtils.getChildText(root, "defaultTimeoutSecond"));
			connectTimeoutSecond = Integer.parseInt(XmlUtils.getChildText(root, "connectTimeoutSecond"));
			dataTimeoutSecond = Integer.parseInt(XmlUtils.getChildText(root, "dataTimeoutSecond"));
			
			host = XmlUtils.getChildText(root, "host");
			port = Integer.parseInt(XmlUtils.getChildText(root, "port"));
			
			user = XmlUtils.getChildText(root, "user");
			password = XmlUtils.getChildText(root, "password");
			
			
			threadPoolSize = Integer.parseInt(XmlUtils.getChildText(root, "threadPoolSize"));
			ftpConnectionSize = Integer.parseInt(XmlUtils.getChildText(root, "ftpConnectionSize"));
			
			rootPath = XmlUtils.getChildText(root, "rootPath");
		catch(Exception e)
			e.printStackTrace();
		
	
    
	

 

package com.ourpalm.resupgrade.util.ftp;

import java.io.File;
import java.io.IOException;
import java.util.List;

/**
 * 客户端
 * @author chuer
 * @date 2015年1月7日 下午2:32:41
 */
public class Client 

	public static void main(String[] args) throws IOException 
		String loalPath = "D:/resource/";
		String ftpPath = "/resource";
		
		File parentFile = new File(loalPath);
		List<UploadResult> resultLists = FtpUtil.upload(ftpPath,parentFile.listFiles());
		
		
		for(UploadResult result : resultLists)
			System.out.println(result);