通过Dynamic proxy 和 Thread local 进行 connection 管理

 

 

public class CassandraAutopubStoreClient implements  InvocationHandler{
	static Logger log = LoggerFactory.getLogger(CassandraAutopubStoreClient.class);
	
	// The next serial number to be assigned
    private static int nextSerialNum = 0;

    private static ThreadLocal<Client> localClient = new ThreadLocal<Client>();

    public static Client getCurrentThreadClient() {
        return localClient.get();
    }

     
	
	private Object store;

    public static Object newInstance(CassandraAutopubStore store) {
	return java.lang.reflect.Proxy.newProxyInstance(
			CassandraAutopubStore.class.getClassLoader(),
			CassandraAutopubStore.class.getInterfaces(),
	    new CassandraAutopubStoreClient(store));
    }

    private CassandraAutopubStoreClient(CassandraAutopubStore store) {
	this.store = store;
    }

	    
	@Override
	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
		 Object result;
			try {
//				if ("getClient".equals(method.getName())){
//					Client client = getCurrentThreadClient();
//					assert client!=null;
//					return client;
//				}else{
					//create client and put the client into threadlocal
					Client client=createClient();
					localClient.set(client);
				    result = method.invoke(store, args);
				    client.getInputProtocol().getTransport().close();
				    localClient.remove();
//				}
		        } catch (InvocationTargetException e) {
			    throw e.getTargetException();
		        } catch (Exception e) {
			    throw new RuntimeException("unexpected invocation exception: " +
						       e.getMessage());
			} finally {
			    System.out.println("after method " + method.getName());
			}
			return result;
	}
	
	public static Client createClient(){
		//TODO: cache the client for reuse property
		TTransport tr = new TSocket("localhost", 9160);
		try {
			tr.open();

		} catch (TTransportException e) {
			throw new StoreException(log, "Cassandra is not running", e);
		}
		log.info("Created CassandraStore connection " + System.identityHashCode(tr));
		final TProtocol proto = new TBinaryProtocol(tr);
		return new Cassandra.Client(proto);
	}

}

 

public StandAloneServer() throws Exception {
		
		System.out.println("Starting REST server on http://"
				+ InetAddress.getLocalHost().getHostAddress() + ":" + PORT
				+ "/");
		component.getServers().add(Protocol.HTTP, PORT);
		component.getDefaultHost().attach(new OpenAutopub());
		component.getDefaultHost().attach("/resources",new StaticResourceApplication());
		component.getClients().add(Protocol.CLAP);
		component.start();
		store = (AutopubStore) CassandraAutopubStoreClient.newInstance(new CassandraAutopubStore(){
			@Override
			public Client getClient() {
				Client client = CassandraAutopubStoreClient.getCurrentThreadClient();
				if (client ==null){
					client = super.getClient();
				}
				return client;
			}
		});
		AutopubResource.setStorage(store);
		
	}