调用示例:
var tran = ThriftPool.Instance().BorrowInstance();
TProtocol protocol = new TBinaryProtocol(tran);
var client = new xxxx(protocol); //xxxx为生成的thrift代理类的Client对象
var ret = client.TestMethod(parameters); //TestMethod为业务方法名称
ThriftPool.Instance().ReturnInstance(tran);
return ret;
配置信息类:
internal class ThriftConfig
{
#region 单例入口 private static ThriftConfig instance = null;
private static object objLock = new object();
public static ThriftConfig Instance()
{
if (instance == null)
{
lock (objLock)
{
if (instance == null)
{
instance = new ThriftConfig();
}
}
}
return instance;
} #endregion #region 私有构造函数 private ThriftConfig()
{
Host = ConfigHelper.GetAppSettingValue("ESHost");
Port = ConfigHelper.GetAppSettingInt32Value("ESPort");
Encode = Encoding.UTF8;
Timeout = 3000;
MaxActive = ConfigHelper.GetAppSettingInt32Value("ESMaxActiveConnectionCount");
MaxIdle = ConfigHelper.GetAppSettingInt32Value("ESMaxIdleConnectionCount");
MinIdle = ConfigHelper.GetAppSettingInt32Value("ESMinIdleConnectionCount");
MaxWait = 5;
} #endregion #region 配置属性定义 public string Host { get; set; } public int Port { get; set; } public Encoding Encode { get; set; } public bool Zipped { get; set; } public int Timeout { get; set; } public int MaxActive { get; set; } public int MaxIdle { get; set; } public int MinIdle { get; set; } public int MaxWait { get; set; }
#endregion }
连接池处理类:
internal class ThriftPool
{
#region 属性 private ThriftConfig config; /// 对象缓存池
private static Stack<TTransport> objectPool { get; set; }
/// 同步对象
private static AutoResetEvent resetEvent; /// 空闲对象数
private static volatile int idleCount = 0; private static volatile int activeCount = 0; /// 同步对象锁
private static object locker = new object(); #endregion #region 单例入口 private static long testcount = 0;
private static ThriftPool instance = null;
private static object objLock = new object();
public static ThriftPool Instance()
{
if (instance == null)
{
lock (objLock)
{
if (instance == null)
{
instance = new ThriftPool();
}
}
}
return instance;
} #endregion #region 构造函数 private ThriftPool()
{
config = ThriftConfig.Instance();
CreateResetEvent();
CreateThriftPool();
} #endregion #region 公有操作方法 /// 从对象池取出一个对象
public TTransport BorrowInstance()
{
lock (locker)
{
//Zkx.Infrastruction.Logger.Log.DebugFormat("借前对象池个数:{0},空闲个数:{1}", objectPool.Count(), idleCount);
TTransport transport;
//对象池无空闲对象 if (idleCount == 0)
{
//对象池已已创建对象数达上限
if (activeCount >= config.MaxActive)
{
//Console.WriteLine("waiting..." + activeCount);
resetEvent.WaitOne();
}
else
{
PushObject(CreateInstance());
}
}
transport = objectPool.Pop();
//Console.WriteLine("Pop 对象..." + transport.IsOpen); //空闲对象数小于最小空闲数,添加一个对象到对象池(已创建数不能超标)
if (--idleCount < config.MinIdle && activeCount < config.MaxActive)
{
PushObject(CreateInstance());
} ValidateInstance(transport);
//Console.WriteLine("借出......对象池个数:{0},空闲个数:{1}," + testcount, objectPool.Count(), idleCount);
return transport;
}
} /// 归还一个对象
/// <param name="instance"></param>
public void ReturnInstance(TTransport instance)
{
lock (locker)
{
// Console.WriteLine("Push 对象..." + instance.IsOpen); //空闲对象数达到上限,不再返回线程池,直接销毁
if (idleCount == config.MaxIdle)
{
DestoryInstance(instance);
}
else
{
ValidateInstance(instance);
PushObject(instance);
//发通知信号,有对象归还到对象池
resetEvent.Set();
//Console.WriteLine("归还...");
}
}
}
#endregion #region 私有方法 /// 创建线程同步对象
private void CreateResetEvent()
{
lock (locker)
{
if (resetEvent == null)
{
resetEvent = new AutoResetEvent(false);
}
}
} /// 创建对象池
private void CreateThriftPool()
{
lock (locker)
{
if (objectPool == null)
{
objectPool = new Stack<TTransport>();
}
}
} /// 添加对象到对象池
private void PushObject(TTransport transport)
{
objectPool.Push(transport);
idleCount++;
} /// 创建一个对象
private TTransport CreateInstance()
{
activeCount++;
var objsocket = new TSocket(config.Host, config.Port);
objsocket.Timeout = 5000;
TTransport transport = objsocket; transport.Open();
//Console.WriteLine("创建对象..." + activeCount);
return transport;
} /// 校验对象
private void ValidateInstance(TTransport instance)
{
if (!instance.IsOpen)
{
//Console.WriteLine("校验_重新打开...");
instance.Open();
}
} /// 销毁对象
private void DestoryInstance(TTransport instance)
{
if (instance.IsOpen)
{
instance.Close();
}
//instance.Flush();
instance.Dispose();
activeCount--;
//Console.WriteLine("销毁...");
} #endregion
}