You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
85 lines
3.1 KiB
85 lines
3.1 KiB
package core.thrift;
|
|
|
|
import org.apache.commons.pool2.impl.GenericObjectPool;
|
|
import org.apache.thrift.TServiceClient;
|
|
import org.apache.thrift.async.TAsyncClient;
|
|
import org.apache.thrift.async.TAsyncClientManager;
|
|
import org.apache.thrift.protocol.TBinaryProtocol;
|
|
import org.apache.thrift.protocol.TProtocol;
|
|
import org.apache.thrift.protocol.TProtocolFactory;
|
|
import org.apache.thrift.transport.TNonblockingSocket;
|
|
import org.apache.thrift.transport.TNonblockingTransport;
|
|
import org.apache.thrift.transport.TSocket;
|
|
import org.apache.thrift.transport.TTransport;
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.stereotype.Component;
|
|
|
|
import javax.annotation.PostConstruct;
|
|
import java.lang.reflect.Constructor;
|
|
import java.util.HashMap;
|
|
import java.util.Map;
|
|
|
|
@Component
|
|
public class SocketConfig {
|
|
@Value("${thrift.ip}")
|
|
private String ip;
|
|
|
|
@Value("${thrift.port}")
|
|
private Integer port;
|
|
|
|
@Value("${thrift.timeout}")
|
|
private Integer timeout;
|
|
|
|
|
|
private GenericObjectPool<TNonblockingSocket> tNonblockingSocketPool;
|
|
|
|
private GenericObjectPool<TSocket> tSocketPool;
|
|
|
|
|
|
private final Map<Class, Constructor> clientMap = new HashMap<>();
|
|
|
|
@PostConstruct
|
|
public void init(){
|
|
tNonblockingSocketPool=new GenericObjectPool<>(new TNonblockingSocketPool(ip,port,timeout));
|
|
tSocketPool=new GenericObjectPool<>(new TSocketPool(ip,port,timeout));
|
|
}
|
|
|
|
public <C extends TAsyncClient> void doExecute(Class<C> c, AsyncClient<C> thrift) throws Exception {
|
|
//从池里获取一个Transport对象
|
|
TNonblockingSocket trans = tNonblockingSocketPool.borrowObject();
|
|
thrift.doExecute(client(c, trans));
|
|
//把一个Transport对象归还到池里
|
|
tNonblockingSocketPool.returnObject(trans);
|
|
}
|
|
|
|
public <C extends TServiceClient> void doExecute(Class<C> c,ServiceClient<C> thrift) throws Exception{
|
|
TSocket trans=tSocketPool.borrowObject();
|
|
thrift.doExecute(client(c,trans));
|
|
tSocketPool.returnObject(trans);
|
|
}
|
|
|
|
private interface Thrift<E> {
|
|
void doExecute(E client) throws Exception;
|
|
}
|
|
|
|
public interface AsyncClient<E extends TAsyncClient> extends Thrift<E>{}
|
|
|
|
public interface ServiceClient<E extends TServiceClient> extends Thrift<E>{}
|
|
|
|
public <T extends TTransport,C> C client(Class<C> c, T trans) throws Exception {
|
|
if (trans instanceof TNonblockingTransport) {
|
|
if (!clientMap.containsKey(c)) {
|
|
clientMap.put(c, c.getConstructor(TProtocolFactory.class, TAsyncClientManager.class, TNonblockingTransport.class));
|
|
}
|
|
TProtocolFactory protocol = new TBinaryProtocol.Factory();
|
|
TAsyncClientManager clientManager = new TAsyncClientManager();
|
|
return (C) clientMap.get(c).newInstance(protocol, clientManager, trans);
|
|
} else {
|
|
if (!clientMap.containsKey(c)) {
|
|
clientMap.put(c, c.getConstructor(TProtocol.class));
|
|
}
|
|
TBinaryProtocol protocol = new TBinaryProtocol(trans);
|
|
return (C) clientMap.get(c).newInstance(protocol);
|
|
}
|
|
}
|
|
}
|
|
|