首页 > 开发 > Java > 正文

Java FTPClient连接池的实现

2024-07-14 08:41:21
字体:
来源:转载
供稿:网友

最近在写一个FTP上传工具,用到了Apache的FTPClient,为了提高上传效率,我采用了多线程的方式,但是每个线程频繁的创建和销毁FTPClient对象势必会造成不必要的开销,因此,此处最好使用一个FTPClient连接池。仔细翻了一下Apache的api,发现它并没有一个FTPClientPool的实现,所以,不得不自己写一个FTPClientPool。下面就大体介绍一下开发连接池的整个过程,供大家参考。

关于对象池

有些对象的创建开销是比较大的,比如数据库连接等。为了减少频繁创建、销毁对象带来的性能消耗,我们可以利用对象池的技术来实现对象的复用。对象池提供了一种机制,它可以管理对象池中对象的生命周期,提供了获取和释放对象的方法,可以让客户端很方便的使用对象池中的对象。

如果我们要自己实现一个对象池,一般需要完成如下功能:

1. 如果池中有可用的对象,对象池应当能返回给客户端
2. 客户端把对象放回池里后,可以对这些对象进行重用
3. 对象池能够创建新的对象来满足客户端不断增长的需求
4. 需要有一个正确关闭池的机制来结束对象的生命周期

Apache的对象池工具包

为了方便我们开发自己的对象池,Apache 提供的common-pool工具包,里面包含了开发通用对象池的一些接口和实现类。其中最基本的两个接口是ObjectPool 和PoolableObjectFactory。

ObjectPool接口中有几个最基本的方法:

1. addObject() : 添加对象到池
2. borrowObject():客户端从池中借出一个对象
3. returnObject():客户端归还一个对象到池中
4. close():关闭对象池,清理内存释放资源等
5. setFactory(ObjectFactory factory):需要一个工厂来制造池中的对象

PoolableObjectFactory接口中几个最基本的方法:

1. makeObject():制造一个对象
2. destoryObject():销毁一个对象
3. validateObject():验证一个对象是否还可用

通过以上两个接口我们就可以自己实现一个对象池了。

实例:开发一个FTPClient对象池

最近在开发一个项目,需要把hdfs中的文件上传到一组ftp服务器,为了提高上传效率,自然考虑到使用多线程的方式进行上传。我上传ftp用的工具是Apache common-net包中的FTPClient,但Apache并没有提供FTPClientPool,于是为了减少FTPClient的创建销毁次数,我们就自己开发一个FTPClientPool来复用FTPClient连接。

通过上面的介绍,我们可以利用Apache提供的common-pool包来协助我们开发连接池。而开发一个简单的对象池,仅需要实现common-pool 包中的ObjectPool和PoolableObjectFactory两个接口即可。下面就看一下我写的实现:

写一个ObjectPool接口的实现FTPClientPool

import java.io.IOException;import java.util.NoSuchElementException;import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.BlockingQueue;import java.util.concurrent.TimeUnit;import org.apache.commons.net.ftp.FTPClient;import org.apache.commons.pool.ObjectPool;import org.apache.commons.pool.PoolableObjectFactory;/*** 实现了一个FTPClient连接池* @author heaven*/public class FTPClientPool implements ObjectPool<FTPClient>{ private static final int DEFAULT_POOL_SIZE = 10; private final BlockingQueue<FTPClient> pool; private final FtpClientFactory factory; /** * 初始化连接池,需要注入一个工厂来提供FTPClient实例 * @param factory * @throws Exception */ public FTPClientPool(FtpClientFactory factory) throws Exception{   this(DEFAULT_POOL_SIZE, factory); } /** * * @param maxPoolSize * @param factory * @throws Exception */ public FTPClientPool(int poolSize, FtpClientFactory factory) throws Exception {   this.factory = factory;   pool = new ArrayBlockingQueue<FTPClient>(poolSize*2);   initPool(poolSize); } /** * 初始化连接池,需要注入一个工厂来提供FTPClient实例 * @param maxPoolSize * @throws Exception */ private void initPool(int maxPoolSize) throws Exception {   for(int i=0;i<maxPoolSize;i++){      //往池中添加对象      addObject();   } } /* (non-Javadoc) * @see org.apache.commons.pool.ObjectPool#borrowObject() */ public FTPClient borrowObject() throws Exception, NoSuchElementException, IllegalStateException {   FTPClient client = pool.take();   if (client == null) {      client = factory.makeObject();      addObject();   }else if(!factory.validateObject(client)){//验证不通过      //使对象在池中失效      invalidateObject(client);      //制造并添加新对象到池中      client = factory.makeObject();      addObject();   }   return client; } /* (non-Javadoc) * @see org.apache.commons.pool.ObjectPool#returnObject(java.lang.Object) */ public void returnObject(FTPClient client) throws Exception {   if ((client != null) && !pool.offer(client,3,TimeUnit.SECONDS)) {      try {        factory.destroyObject(client);      } catch (IOException e) {        e.printStackTrace();      }   } } public void invalidateObject(FTPClient client) throws Exception {   //移除无效的客户端   pool.remove(client); } /* (non-Javadoc) * @see org.apache.commons.pool.ObjectPool#addObject() */ public void addObject() throws Exception, IllegalStateException, UnsupportedOperationException {   //插入对象到队列   pool.offer(factory.makeObject(),3,TimeUnit.SECONDS); } public int getNumIdle() throws UnsupportedOperationException {   return 0; } public int getNumActive() throws UnsupportedOperationException {   return 0; } public void clear() throws Exception, UnsupportedOperationException { } /* (non-Javadoc) * @see org.apache.commons.pool.ObjectPool#close() */ public void close() throws Exception {   while(pool.iterator().hasNext()){      FTPClient client = pool.take();      factory.destroyObject(client);   } } public void setFactory(PoolableObjectFactory<FTPClient> factory) throws IllegalStateException, UnsupportedOperationException { }}

再写一个PoolableObjectFactory接口的实现FTPClientFactory

import java.io.IOException;import org.apache.commons.net.ftp.FTPClient;import org.apache.commons.net.ftp.FTPReply;import org.apache.commons.pool.PoolableObjectFactory;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import com.hdfstoftp.util.FTPClientException;/*** FTPClient工厂类,通过FTPClient工厂提供FTPClient实例的创建和销毁* @author heaven*/public class FtpClientFactory implements PoolableObjectFactory<FTPClient> {private static Logger logger = LoggerFactory.getLogger("file"); private FTPClientConfigure config; //给工厂传入一个参数对象,方便配置FTPClient的相关参数 public FtpClientFactory(FTPClientConfigure config){   this.config=config; } /* (non-Javadoc) * @see org.apache.commons.pool.PoolableObjectFactory#makeObject() */ public FTPClient makeObject() throws Exception {   FTPClient ftpClient = new FTPClient();   ftpClient.setConnectTimeout(config.getClientTimeout());   try {      ftpClient.connect(config.getHost(), config.getPort());      int reply = ftpClient.getReplyCode();      if (!FTPReply.isPositiveCompletion(reply)) {        ftpClient.disconnect();        logger.warn("FTPServer refused connection");        return null;      }      boolean result = ftpClient.login(config.getUsername(), config.getPassword());      if (!result) {        throw new FTPClientException("ftpClient登陆失败! userName:" + config.getUsername() + " ; password:" + config.getPassword());      }      ftpClient.setFileType(config.getTransferFileType());      ftpClient.setBufferSize(1024);      ftpClient.setControlEncoding(config.getEncoding());      if (config.getPassiveMode().equals("true")) {        ftpClient.enterLocalPassiveMode();      }   } catch (IOException e) {      e.printStackTrace();   } catch (FTPClientException e) {      e.printStackTrace();   }   return ftpClient; } /* (non-Javadoc) * @see org.apache.commons.pool.PoolableObjectFactory#destroyObject(java.lang.Object) */ public void destroyObject(FTPClient ftpClient) throws Exception {   try {      if (ftpClient != null && ftpClient.isConnected()) {        ftpClient.logout();      }   } catch (IOException io) {      io.printStackTrace();   } finally {      // 注意,一定要在finally代码中断开连接,否则会导致占用ftp连接情况      try {        ftpClient.disconnect();      } catch (IOException io) {        io.printStackTrace();      }   } } /* (non-Javadoc) * @see org.apache.commons.pool.PoolableObjectFactory#validateObject(java.lang.Object) */ public boolean validateObject(FTPClient ftpClient) {   try {      return ftpClient.sendNoOp();   } catch (IOException e) {      throw new RuntimeException("Failed to validate client: " + e, e);   } } public void activateObject(FTPClient ftpClient) throws Exception { } public void passivateObject(FTPClient ftpClient) throws Exception { }}

最后,我们最好给工厂传递一个参数对象,方便我们设置FTPClient的一些参数

package org.apache.commons.pool.impl.contrib;/** * FTPClient配置类,封装了FTPClient的相关配置 * * @author heaven */public class FTPClientConfigure { private String host; private int port; private String username; private String password; private String passiveMode; private String encoding; private int clientTimeout; private int threadNum; private int transferFileType; private boolean renameUploaded; private int retryTimes; public String getHost() {     return host; } public void setHost(String host) {     this. host = host; } public int getPort() {     return port; } public void setPort(int port) {     this. port = port; } public String getUsername() {     return username; } public void setUsername(String username) {     this. username = username; } public String getPassword() {     return password; } public void setPassword(String password) {     this. password = password; } public String getPassiveMode() {     return passiveMode; } public void setPassiveMode(String passiveMode) {     this. passiveMode = passiveMode; } public String getEncoding() {     return encoding; } public void setEncoding(String encoding) {     this. encoding = encoding; } public int getClientTimeout() {     return clientTimeout; } public void setClientTimeout( int clientTimeout) {     this. clientTimeout = clientTimeout; } public int getThreadNum() {     return threadNum; } public void setThreadNum( int threadNum) {     this. threadNum = threadNum; } public int getTransferFileType() {     return transferFileType; } public void setTransferFileType( int transferFileType) {     this. transferFileType = transferFileType; } public boolean isRenameUploaded() {     return renameUploaded; } public void setRenameUploaded( boolean renameUploaded) {     this. renameUploaded = renameUploaded; } public int getRetryTimes() {     return retryTimes; } public void setRetryTimes( int retryTimes) {     this. retryTimes = retryTimes; } @Override public String toString() {     return "FTPClientConfig [host=" + host + "/n port=" + port + "/n username=" + username + "/n password=" + password  + "/n passiveMode=" + passiveMode          + "/n encoding=" + encoding + "/n clientTimeout=" + clientTimeout + "/n threadNum=" + threadNum + "/n transferFileType="          + transferFileType + "/n renameUploaded=" + renameUploaded + "/n retryTimes=" + retryTimes + "]" ; }}

 FTPClientPool连接池类管理FTPClient对象的生命周期,负责对象的借出、规划、池的销毁等;FTPClientPool类依赖于FtpClientFactory类,由这个工程类来制造和销毁对象;FtpClientFactory又依赖FTPClientConfigure类,FTPClientConfigure负责封装FTPClient的配置参数。至此,我们的FTPClient连接池就开发完成了。

需要注意的是,FTPClientPool中用到了一个阻塞队列ArrayBlockingQueue来管理存放FTPClient对象,关于阻塞队列,请参考我的这篇文章: 【Java并发之】BlockingQueue

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持VeVb武林网。


注:相关教程知识阅读请移步到JAVA教程频道。
发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表