Java多线程实现FTP批量上传文件
更新时间：2022年06月23日 11:45:37 作者：高级软件开发工程师
这篇文章主要为大家详细介绍了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();
}
}
/**
* 初始化连接
*
* @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，所以目录名或文件名需要转码
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。
相关文章
这篇文章主要介绍了Spring 异常单元测试的解决，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-06-06
在web开发中我们都要进行前端传参后端取参的过程,本文主要介绍了SpringMVC前后端参数映射,针对GET, POST, PUT, DELETE 请求的参数该如何映射,感兴趣的可以了解一下2023-08-08
Nacos 是阿里巴巴推出来的一个新开源项目，这是一个更易于构建云原生应用的动态服务发现、配置管理和服务管理平台，在项目开发过程中，我们经常使用nacos作为配置中心和注册中心。本文章我们就从代码层面研究下springboot是如何整合nacos使用的2022-07-07
这篇文章主要介绍了Java类和对象的设计原理，文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的朋友可以参考一下2022-07-07
这篇文章主要介绍了Java中的信号量Semaphore详细解读,Java信号量机制可以用来保证线程互斥,创建Semaphore对象传入一个整形参数,类似于公共资源,需要的朋友可以参考下2023-11-11
java新增关联的三张表,每张表要求都插入集合,代码实现方式
这篇文章主要介绍了java新增关联的三张表,每张表要求都插入集合,代码实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2023-12-12
这篇文章主要给大家介绍了关于Spring Boot 2.0快速构建服务组件的相关资料，文中通过示例代码介绍的非常详细，对大家学习或者使用Spring Boot 2.0具有一定的参考学习价值，需要的朋友们下面来一起学习学习吧2019-04-04
在Java开发中,集合（Collection）是一个重要的数据结构,广泛应用于各种场景,计算集合中的组内平均值是一个常见的操作,本文将深入探讨如何使用Java来计算集合中的组内平均值,涵盖基本概念、具体实现、优化策略和实用示例,需要的朋友可以参考下2024-06-06
SpringBoot整合Javamail实现邮件发送的详细过程
日常开发过程中,我们经常需要使用到邮件发送任务,比方说验证码的发送、日常信息的通知等,下面这篇文章主要给大家介绍了关于SpringBoot整合Javamail实现邮件发送的详细过程,需要的朋友可以参考下2022-10-10
这篇文章主要为大家详细介绍了Java网络编程UDP实现消息发送及聊天，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2021-07-07
最新评论