package com.chu.cassandratest;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.thrift.transport.TTransportException;
/**
* cassandra连接池
* @author chuer
* @date 2014年12月31日 上午10:05:26
*/
public class CassandraConnectionPool {
Semaphore access = null;
CassandraConnection[] pool = null;
boolean[] used = null;
int round = 0;
int conn_num = 0;
public CassandraConnectionPool(int conn_num) {
this.conn_num = conn_num;
init();
}
private void init() {
access = new Semaphore(conn_num);//有几个连接就允许有几个线程同时访问连接池。
pool = new CassandraConnection[conn_num];
used = new boolean[conn_num];
for (int i = 0; i < pool.length; i++) {
try {
pool[i] = new CassandraConnection(CassandraConnection.HOST);
} catch (TTransportException e) {
e.printStackTrace();
}
}
}
public CassandraConnection getConn() throws InterruptedException {
if (access.tryAcquire(3, TimeUnit.SECONDS)) {
synchronized (this) {
for (int i = 0; i < pool.length; i++) {
if (!used[i]) {
used[i] = true;
return pool[i];
}
}
}
}
throw new RuntimeException("all client is too busy");
}
public void releaseConn(CassandraConnection client) {
boolean released = false;
synchronized (this) {
for (int i = 0; i < pool.length; i++) {
if (client == pool[i] && used[i]) {
used[i] = false;
released = true;
break;
}
}
}
if (released)
access.release();
}
public void shutdownPool() {
if (pool != null) {
for (int i = 0; i < pool.length; i++){
pool[i].close();
}
}
}
}
本文代码，摘抄修改于网上的另一篇文章，网络在哪里了，恕不能注明出处。
package com.chu.cassandratest;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
/**
* cassandra连接
* @author chuer
* @date 2014年12月31日 上午10:09:47
*/
public class CassandraConnection {
private Cassandra.Client client = null;
private TTransport tr = null;
private String host = null;
public final static String HOST = "127.0.0.1";
public CassandraConnection(String host) throws TTransportException {
this.host = host;
this.connect(host);
}
private synchronized void connect(String host) throws TTransportException {
tr = new TSocket(host, 9160);
TProtocol proto = new TBinaryProtocol(tr);
client = new Cassandra.Client(proto);
tr.open();
}
public synchronized void close() {
if (tr != null && tr.isOpen())
tr.close();
}
public Cassandra.Client getClient() {
return client;
}
public String getHost() {
return host;
}
}
package com.chu.cassandratest;
import org.apache.cassandra.thrift.Cassandra.Client;
public class TestCassandraPool {
public static void main(String[] args)throws Exception {
CassandraConnectionPool pool = new CassandraConnectionPool(10);
CassandraConnection conn1 = pool.getConn();
CassandraConnection conn2 = pool.getConn();
CassandraConnection conn3 = pool.getConn();
CassandraConnection conn4 = pool.getConn();
CassandraConnection conn5 = pool.getConn();
Client client1 = conn1.getClient();
Client client2 = conn2.getClient();
Client client3 = conn3.getClient();
Client client4 = conn4.getClient();
Client client5 = conn5.getClient();
System.out.println(client1);
System.out.println(client2);
System.out.println(client3);
System.out.println(client4);
System.out.println(client5);
pool.releaseConn(conn1);
pool.releaseConn(conn2);
pool.releaseConn(conn3);
pool.releaseConn(conn4);
pool.releaseConn(conn5);
}
}
输出如下：
org.apache.cassandra.thrift.Cassandra$Client@1c160cb
org.apache.cassandra.thrift.Cassandra$Client@b9b67b
org.apache.cassandra.thrift.Cassandra$Client@7b4ed7
org.apache.cassandra.thrift.Cassandra$Client@17535b6
org.apache.cassandra.thrift.Cassandra$Client@1979eb
、replaceFirst()和replaceAll()区别)
)
