Java多线程实现FTP批量上传文件
本文实例为大家分享了Java多线程实现FTP批量上传文件的具体代码，供大家参考，具体内容如下
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();
}
}
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;
}
}
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;
}
public void disconnect() throws IOException {
if (ftp.isConnected()) {
try {
ftp.loGout();
ftp.disconnect();
is_connected = false;
} catch (IOException ex) {
ex.printStackTrace();
}
}
}
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;
}
public boolean isConnected() {
return is_connected;
}
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());
}
}
public FtpConnection getFtp() {
FtpConnection poll = null;
try {
//取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 Blocking 有新的对象被加入为止
poll = arrayBlockingQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
return poll;
}
public boolean relase(FtpConnection ftp){
return arrayBlockingQueue.offer(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;
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;
public class FtpUtil {
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，所以目录名或文件名需要转码
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
相关文章