Java FTPClient连接池的实现
最近在写一个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
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
上一篇: 磨刀不误砍柴工---PHP开发工具集锦