Thrift连接池实现.docx
《Thrift连接池实现.docx》由会员分享，可在线阅读，更多相关《Thrift连接池实现.docx（25页珍藏版）》请在冰点文库上搜索。
Thrift连接池实现
简介
Thrift是Facebook的核心框架之一，使不同的开发语言开发的系统可以通过该框架实现彼此的通信，类似于webservice，但是Thrift提供了近乎变态的效率和开发的方便性，是webservice所不能比拟的。
给分布式开发带来了极大的方便。
但是这柄利器也有一些不完美。
问题
首先文档相当的少，只有一个wiki网站提供相应的帮助。
这对于Thrift的推广极为不利。
其次框架本身实现有一些缺陷，就Thrift的java部分来说，没有提供连接池的支持，对RPC的调用效率有所影响。
对于文档稀少的问题，只能是通过一些Thrift的开发者和使用者多供献一些自己的心得来解决。
这得需要一个过程。
而连接池的问题的解决则可以快速一些。
提到池一般做过Java开发的肯定会想到ObjectPool，ApacheCommons项目确实给我们的开发得来了很大的便利性，其中的pool项目正是我们实现thrift连接池的基础，当然也少不了神器springframework。
实现
一，定义thrift连接池接口
ConnectionProvider
/*
*@（#）ConnectionProvider.java0.105/11/17
*
*Copyright2010QISI,Inc.Allrightsreserved.
*QISIPROPRIETARY/CONFIDENTIAL.Useissubjecttolicenseterms.
*/
packagecom.qidea.thrift.pool;
importorg.apache.thrift.transport.TSocket;
/**
*
*@authorsunwei
*@version2010-8-6
*@sinceJDK1.5
*/
publicinterfaceConnectionProvider
{
/**
*取链接池中的一个链接
*
*@return
*/
publicTSocketgetConnection（）;
/**
*返回链接
*
*@paramsocket
*/
publicvoidreturnCon（TSocketsocket）;
}
二，实现连接池
GenericConnectionProvider
/*
*@（#）DefaultConnectionProviderImpl.java0.105/11/17
*
*Copyright2010QISI,Inc.Allrightsreserved.
*QISIPROPRIETARY/CONFIDENTIAL.Useissubjecttolicenseterms.
*/
packagecom.qidea.thrift.pool;
importmons.pool.ObjectPool;
importmons.pool.impl.GenericObjectPool;
importorg.apache.thrift.transport.TSocket;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
importorg.springframework.beans.factory.DisposableBean;
importorg.springframework.beans.factory.InitializingBean;
/**
*
*@authorsunwei
*@version2010-8-10
*@sinceJDK1.5
*/
publicclassGenericConnectionProviderimplementsConnectionProvider,
InitializingBean,DisposableBean
{
publicstaticfinalLoggerlogger=LoggerFactory
.getLogger（GenericConnectionProvider.class）;
/**服务的IP地址*/
privateStringserviceIP;
/**服务的端口*/
privateintservicePort;
/**连接超时配置*/
privateintconTimeOut;
/**可以从缓存池中分配对象的最大数量*/
privateintmaxActive=GenericObjectPool.DEFAULT_MAX_ACTIVE;
/**缓存池中最大空闲对象数量*/
privateintmaxIdle=GenericObjectPool.DEFAULT_MAX_IDLE;
/**缓存池中最小空闲对象数量*/
privateintminIdle=GenericObjectPool.DEFAULT_MIN_IDLE;
/**阻塞的最大数量*/
privatelongmaxWait=GenericObjectPool.DEFAULT_MAX_WAIT;
/**从缓存池中分配对象，是否执行PoolableObjectFactory.validateObject方法*/
privatebooleantestOnBorrow=GenericObjectPool.DEFAULT_TEST_ON_BORROW;
privatebooleantestOnReturn=GenericObjectPool.DEFAULT_TEST_ON_RETURN;
privatebooleantestWhileIdle=GenericObjectPool.DEFAULT_TEST_WHILE_IDLE;
/**对象缓存池*/
privateObjectPoolobjectPool=null;
/**
*
*/
@Override
publicvoidafterPropertiesSet（）throwsException
{
//对象池
objectPool=newGenericObjectPool（）;
//
（（GenericObjectPool）objectPool）.setMaxActive（maxActive）;
（（GenericObjectPool）objectPool）.setMaxIdle（maxIdle）;
（（GenericObjectPool）objectPool）.setMinIdle（minIdle）;
（（GenericObjectPool）objectPool）.setMaxWait（maxWait）;
（（GenericObjectPool）objectPool）.setTestOnBorrow（testOnBorrow）;
（（GenericObjectPool）objectPool）.setTestOnReturn（testOnReturn）;
（（GenericObjectPool）objectPool）.setTestWhileIdle（testWhileIdle）;
（（GenericObjectPool）objectPool）
.setWhenExhaustedAction（GenericObjectPool.WHEN_EXHAUSTED_BLOCK）;
//设置factory
ThriftPoolableObjectFactorythriftPoolableObjectFactory=newThriftPoolableObjectFactory（
serviceIP,servicePort,conTimeOut）;
objectPool.setFactory（thriftPoolableObjectFactory）;
}
@Override
publicvoiddestroy（）
{
try
{
objectPool.close（）;
}
catch（Exceptione）
{
thrownewRuntimeException（"erorrdestroy（）",e）;
}
}
@Override
publicTSocketgetConnection（）
{
try
{
TSocketsocket=（TSocket）objectPool.borrowObject（）;
returnsocket;
}
catch（Exceptione）
{
thrownewRuntimeException（"errorgetConnection（）",e）;
}
}
@Override
publicvoidreturnCon（TSocketsocket）
{
try
{
objectPool.returnObject（socket）;
}
catch（Exceptione）
{
thrownewRuntimeException（"errorreturnCon（）",e）;
}
}
publicStringgetServiceIP（）
{
returnserviceIP;
}
publicvoidsetServiceIP（StringserviceIP）
{
this.serviceIP=serviceIP;
}
publicintgetServicePort（）
{
returnservicePort;
}
publicvoidsetServicePort（intservicePort）
{
this.servicePort=servicePort;
}
publicintgetConTimeOut（）
{
returnconTimeOut;
}
publicvoidsetConTimeOut（intconTimeOut）
{
this.conTimeOut=conTimeOut;
}
publicintgetMaxActive（）
{
returnmaxActive;
}
publicvoidsetMaxActive（intmaxActive）
{
this.maxActive=maxActive;
}
publicintgetMaxIdle（）
{
returnmaxIdle;
}
publicvoidsetMaxIdle（intmaxIdle）
{
this.maxIdle=maxIdle;
}
publicintgetMinIdle（）
{
returnminIdle;
}
publicvoidsetMinIdle（intminIdle）
{
this.minIdle=minIdle;
}
publiclonggetMaxWait（）
{
returnmaxWait;
}
publicvoidsetMaxWait（longmaxWait）
{
this.maxWait=maxWait;
}
publicbooleanisTestOnBorrow（）
{
returntestOnBorrow;
}
publicvoidsetTestOnBorrow（booleantestOnBorrow）
{
this.testOnBorrow=testOnBorrow;
}
publicbooleanisTestOnReturn（）
{
returntestOnReturn;
}
publicvoidsetTestOnReturn（booleantestOnReturn）
{
this.testOnReturn=testOnReturn;
}
publicbooleanisTestWhileIdle（）
{
returntestWhileIdle;
}
publicvoidsetTestWhileIdle（booleantestWhileIdle）
{
this.testWhileIdle=testWhileIdle;
}
publicObjectPoolgetObjectPool（）
{
returnobjectPool;
}
publicvoidsetObjectPool（ObjectPoolobjectPool）
{
this.objectPool=objectPool;
}
}
ThriftPoolableObjectFactory
/*
*@（#）ThriftPoolableObjectFactory.java0.105/11/17
*
*Copyright2010QISI,Inc.Allrightsreserved.
*QISIPROPRIETARY/CONFIDENTIAL.Useissubjecttolicenseterms.
*/
packagecom.qidea.thrift.pool;
importmons.pool.PoolableObjectFactory;
importorg.apache.thrift.transport.TSocket;
importorg.apache.thrift.transport.TTransport;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
/**
*
*@authorsunwei
*@version2010-8-10
*@sinceJDK1.5
*/
publicclassThriftPoolableObjectFactoryimplementsPoolableObjectFactory
{
/**日志记录器*/
publicstaticfinalLoggerlogger=LoggerFactory
.getLogger（ThriftPoolableObjectFactory.class）;
/**服务的IP*/
privateStringserviceIP;
/**服务的端口*/
privateintservicePort;
/**超时设置*/
privateinttimeOut;
/**
*
*@paramserviceIP
*@paramservicePort
*@paramtimeOut
*/
publicThriftPoolableObjectFactory（StringserviceIP,intservicePort,
inttimeOut）
{
this.serviceIP=serviceIP;
this.servicePort=servicePort;
this.timeOut=timeOut;
}
@Override
publicvoiddestroyObject（Objectarg0）throwsException
{
if（arg0instanceofTSocket）
{
TSocketsocket=（TSocket）arg0;
if（socket.isOpen（））
{
socket.close（）;
}
}
}
/**
*
*/
@Override
publicObjectmakeObject（）throwsException
{
try
{
TTransporttransport=newTSocket（this.serviceIP,
this.servicePort,this.timeOut）;
transport.open（）;
returntransport;
}
catch（Exceptione）
{
logger.error（"errorThriftPoolableObjectFactory（）",e）;
thrownewRuntimeException（e）;
}
}
@Override
publicbooleanvalidateObject（Objectarg0）
{
try
{
if（arg0instanceofTSocket）
{
TSocketthriftSocket=（TSocket）arg0;
if（thriftSocket.isOpen（））
{
returntrue;
}
else
{
returnfalse;
}
}
else
{
returnfalse;
}
}
catch（Exceptione）
{
returnfalse;
}
}
@Override
publicvoidpassivateObject（Objectarg0）throwsException
{
//DONOTHING
}
@Override
publicvoidactivateObject（Objectarg0）throwsException
{
//DONOTHING
}
publicStringgetServiceIP（）
{
returnserviceIP;
}
publicvoidsetServiceIP（StringserviceIP）
{
this.serviceIP=serviceIP;
}
publicintgetServicePort（）
{
returnservicePort;
}
publicvoidsetServicePort（intservicePort）
{
this.servicePort=servicePort;
}
publicintgetTimeOut（）
{
returntimeOut;
}
publicvoidsetTimeOut（inttimeOut）
{
this.timeOut=timeOut;
}
}
三，定义连接的管理类
ConnectionManager
/*
*@（#）ConnectionManager.java0.105/11/17
*
*Copyright2010QISI,Inc.Allrightsreserved.
*QISIPROPRIETARY/CONFIDENTIAL.Useissubjecttolicenseterms.
*/
packagecom.qidea.thrift.pool;
importorg.aopalliance.intercept.MethodInterceptor;
importorg.aopalliance.intercept.MethodInvocation;
importorg.apache.thrift.transport.TSocket;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
/**
*
*@authorsunwei
*@version2010-8-10
*@sinceJDK1.5
*/
publicclassConnectionManagerimplementsMethodInterceptor
{
/**日志记录器*/
publicLoggerlogger=LoggerFactory.getLogger（ConnectionManager.class）;
/**保存local对象*/
ThreadLocalsocketThreadSafe=newThreadLocal（）;
/**连接提供池*/
publicConnectionProviderconnectionProvider;
@Override
publicObjectinvoke（MethodInvocationarg0）throwsThrowable
{
TSocketsocket=null;
try
{
socket=connectionProvider.getConnection（）;
socketThreadSafe.set（socket）;
Objectret=arg0.proceed（）;
returnret;
}
catch（Exceptione）
{
logger.error（"errorConnectionManager.invoke（）",e）;
thrownewException（e）;
}
finally
{
connectionProvider.returnCon（socket）;
socketThreadSafe.remove（）;
}
}
/**
*取socket
*
*@return
*/
publicTSocketgetSocket（）
{
returnsocketThreadSafe.get（）;
}
publicConnectionProvidergetConnectionProvider（）
{
returnconnectionProvider;
}
publicvoidsetConnectionProvider（ConnectionProviderconnectionProvider）
{
this.connectionProvider=connectionProvider;
}
}
四，定义spring配置，对受管的bean提供thrift连接
Thrift连接池spring配置
xmlversion="1.0"encoding="UTF-8"?
>
//www.springframework.org/schema/beans"
xmlns:
xsi="http:
//www.w3.org/2001/XMLSchema-instance"xmlns:
jee="http:
//www.springframework.org/schema/jee"
xmlns:
aop="http:
//www.springframework.org/schema/aop"xmlns:
tx="http:
//www.springframework.org/schema/tx"
xmlns:
flex="http:
//www.springframework.org/schema/flex"x
升级会员 