Thrift提供的网络服务模型：单线程、多线程、事件驱动，从另一个角度划分为：阻塞服务模型、非阻塞服务模型。
-
阻塞服务模型：TSimpleServer、TThreadPoolServer。
-
非阻塞服务模型：TNonblockingServer、THsHaServer和TThreadedSelectorServer。
TThreadPoolServer类图
TThreadPoolServer模式采用阻塞socket方式工作，主线程负责阻塞式监听是否有新socket到来，每当新的客户端连接请求过来，就将其封装起来，然后交给线程池来处理。线程池来完成具体的业务处理，将结果发给客户端。流程图如下所示。
TThreadPoolServer
我们先来看看Args 类，Args 类多了几个陌生的参数，
public static class Args extends AbstractServerArgs<Args> {
//用来设置线程池的时候使用
public int minWorkerThreads = 5;
public int maxWorkerThreads = Integer.MAX_VALUE;
// 线程池
public ExecutorService executorService;
// 用来设置线程池shutdown后，主线程等待多长时间
public int stopTimeoutVal = 60;
public TimeUnit stopTimeoutUnit = TimeUnit.SECONDS;
/**
* requestTimeout和beBackoffSlotLength共同用来设置这次重试和上次重试相隔时常，详细的方式下面有讲解
*/
// 用来设置总共重试时长
public int requestTimeout = 20;
public TimeUnit requestTimeoutUnit = TimeUnit.SECONDS;
// 这个参数用来设置重试步伐的
public int beBackoffSlotLength = 100;
public TimeUnit beBackoffSlotLengthUnit = TimeUnit.MILLISECONDS;
public Args(TServerTransport transport) {
super(transport);
}
public Args minWorkerThreads(int n) {
minWorkerThreads = n;
return this;
}
public Args maxWorkerThreads(int n) {
maxWorkerThreads = n;
return this;
}
public Args stopTimeoutVal(int n) {
stopTimeoutVal = n;
return this;
}
public Args stopTimeoutUnit(TimeUnit tu) {
stopTimeoutUnit = tu;
return this;
}
public Args requestTimeout(int n) {
requestTimeout = n;
return this;
}
public Args requestTimeoutUnit(TimeUnit tu) {
requestTimeoutUnit = tu;
return this;
}
//Binary exponential backoff slot length
public Args beBackoffSlotLength(int n) {
beBackoffSlotLength = n;
return this;
}
//Binary exponential backoff slot time unit
public Args beBackoffSlotLengthUnit(TimeUnit tu) {
beBackoffSlotLengthUnit = tu;
return this;
}
public Args executorService(ExecutorService executorService) {
this.executorService = executorService;
return this;
}
}
我们在看下TThreadPoolServer的源码，这个是TThreadPoolServer的构造函数，在构造函数里实例化线程池。
public TThreadPoolServer(Args args) {
super(args);
stopTimeoutUnit = args.stopTimeoutUnit;
stopTimeoutVal = args.stopTimeoutVal;
requestTimeoutUnit = args.requestTimeoutUnit;
requestTimeout = args.requestTimeout;
beBackoffSlotInMillis = args.beBackoffSlotLengthUnit.toMillis(args.beBackoffSlotLength);
// 实例化线程池 可以自己设计线程池后传进来，或者ThreadPoolServer给你创建
executorService_ = args.executorService != null ?
args.executorService : createDefaultExecutorService(args);
}
// 创建线程池代码
private static ExecutorService createDefaultExecutorService(Args args) {
// 线程池等待队列 此队列中不缓存任何一个任务。向线程池提交任务时，如果没有空闲线程来运行任务，
// 则入列操作会阻塞。当有线程来获取任务时，出列操作会唤醒执行入列操作的线程。
// 从这个特性来看，SynchronousQueue是一个无界队列，因此当使用SynchronousQueue作为线程池的阻塞队列时，
// 参数maximumPoolSizes没有任何作用。
SynchronousQueue<Runnable> executorQueue =
new SynchronousQueue<Runnable>();
return new ThreadPoolExecutor(args.minWorkerThreads,
args.maxWorkerThreads,
args.stopTimeoutVal,
args.stopTimeoutUnit,
executorQueue);
}
serve()函数
- preServe()函数：开启服务器进行监听
- execute()函数：将处理客户端请求交给线程池
- waitForShutdown()函数：服务端停止工作后，关闭线程池
// 重头戏，通过serve()启动服务端
public void serve() {
// 服务器进行监听
if (!preServe()) {
return;
}
// 在execute()函数里面获取新的客户端的连接请求。然后交给线程池进行相应的业务处理
execute();
// 服务端停止工作后，通过这个函数关闭线程池
waitForShutdown();
setServing(false);
}
- preServe()函数，用户来开启服务端对客户端的监听
protected boolean preServe() {
try {
// 服务器进行监听
serverTransport_.listen();
} catch (TTransportException ttx) {
LOGGER.error("Error occurred during listening.", ttx);
return false;
}
// Run the preServe event
if (eventHandler_ != null) {
eventHandler_.preServe();
}
stopped_ = false;
setServing(true);
return true;
}
execute()函数，服务端接收到客户端连接的请求后，将其封装成WorkerProcess类，丢给线程池
protected void execute() {
int failureCount = 0;
// stopped_ 是服务端停止的标记
while (!stopped_) {
try {
// 接收到了来自新的客户端的连接请求，套接字
TTransport client = serverTransport_.accept();
// WorkerProcess类继承了Runnable类，将客户端封装在里面，好扔给线程池处理
// 对客户端详细的处理过程在WorkerProcess类的run()方法里
WorkerProcess wp = new WorkerProcess(client);
// 记录加入线程池的重试次数
int retryCount = 0;
// 剩余的重试时间 刚开始等于requestTimeout
long remainTimeInMillis = requestTimeoutUnit.toMillis(requestTimeout);
while (true) {
try {
// 交给线程池处理
executorService_.execute(wp);
break;
} catch (Throwable t) {
// 如果报错 根据错误类型进行重试
if (t instanceof RejectedExecutionException) {
// 加入线程池被拒绝了
// 重试次数加一
retryCount++;
try {
if (remainTimeInMillis > 0) {
//do a truncated 20 binary exponential backoff sleep
// sleepTimeInMillis: 每次重试失败后，休眠的时长，
long sleepTimeInMillis = ((long) (random.nextDouble() *
(1L << Math.min(retryCount, 20)))) * beBackoffSlotInMillis;
// 获取sleepTimeInMillis和remainTimeInMillis较小的值
sleepTimeInMillis = Math.min(sleepTimeInMillis, remainTimeInMillis);
TimeUnit.MILLISECONDS.sleep(sleepTimeInMillis);
// 减去这次休眠的时间
remainTimeInMillis = remainTimeInMillis - sleepTimeInMillis;
} else {
client.close();
wp = null;
LOGGER.warn("Task has been rejected by ExecutorService " + retryCount
+ " times till timedout, reason: " + t);
break;
}
} catch (InterruptedException e) {
LOGGER.warn("Interrupted while waiting to place client on executor queue.");
Thread.currentThread().interrupt();
break;
}
} else if (t instanceof Error) {
LOGGER.error("ExecutorService threw error: " + t, t);
throw (Error) t;
} else {
//for other possible runtime errors from ExecutorService, should also not kill serve
LOGGER.warn("ExecutorService threw error: " + t, t);
break;
}
}
}
} catch (TTransportException ttx) {
if (!stopped_) {
// 加入现场失败次数加一
++failureCount;
LOGGER.warn("Transport error occurred during acceptance of message.", ttx);
}
}
}
}
- waitForShutdown()函数，当服务端线程停止后， 停止线程池
protected void waitForShutdown() {
// 不在接受新的线程，并且等待之前提交的线程都执行完在关闭，
executorService_.shutdown();
// Loop until awaitTermination finally does return without a interrupted
// exception. If we don't do this, then we'll shut down prematurely. We want
// to let the executorService clear it's task queue, closing client sockets
// appropriately.
/**
* 循环执行知道调用awaitTermination() 后不抛出异常，如果不这样做线程池会关闭的过早
* 我们希望线程池可以让自己等待队列里的任务也执行完毕，然后再关闭于客户端的socket连接
* 就是再线程池执行shutdown()方法后，留stopTimeoutVal长的时间执行完等待队列里的任务。
*/
long timeoutMS = stopTimeoutUnit.toMillis(stopTimeoutVal);
long now = System.currentTimeMillis();
while (timeoutMS >= 0) {
try {
// 该方法调用会被阻塞，直到所有任务执行完毕并且shutdown请求被调用，
// 或者参数中定义的timeout时间到达或者当前线程被打断，
// 这几种情况任意一个发生了就会导致该方法的执行。
executorService_.awaitTermination(timeoutMS, TimeUnit.MILLISECONDS);
break;
} catch (InterruptedException ix) {
long newnow = System.currentTimeMillis();
timeoutMS -= (newnow - now);
now = newnow;
}
}
}
WorkerProcess 类源码，WorkerProcess 类继承了Runnable ，在run()函数里进行相应的业务逻辑处理。
private class WorkerProcess implements Runnable {
/**
* Client that this services.
*/
private TTransport client_;
/**
* Default constructor.
*
* @param client Transport to process
*/
private WorkerProcess(TTransport client) {
client_ = client;
}
/**
* Loops on processing a client forever
*/
public void run() {
// 业务逻辑处理器
TProcessor processor = null;
// 传输层
TTransport inputTransport = null;
TTransport outputTransport = null;
// 协议层
TProtocol inputProtocol = null;
TProtocol outputProtocol = null;
TServerEventHandler eventHandler = null;
ServerContext connectionContext = null;
try {
// 获取客户端相应的处理器，传输层，协议层
processor = processorFactory_.getProcessor(client_);
inputTransport = inputTransportFactory_.getTransport(client_);
outputTransport = outputTransportFactory_.getTransport(client_);
inputProtocol = inputProtocolFactory_.getProtocol(inputTransport);
outputProtocol = outputProtocolFactory_.getProtocol(outputTransport);
eventHandler = getEventHandler();
if (eventHandler != null) {
connectionContext = eventHandler.createContext(inputProtocol, outputProtocol);
}
// we check stopped_ first to make sure we're not supposed to be shutting
// down. this is necessary for graceful shutdown.
while (true) {
if (eventHandler != null) {
eventHandler.processContext(connectionContext, inputTransport, outputTransport);
}
if (stopped_) {
break;
}
// 进行业务逻辑处理，如果处理完一个请求以后，下一个请求还没来
// 那么这个线程将会阻塞在这里
processor.process(inputProtocol, outputProtocol);
}
} catch (Exception x) {
// We'll usually receive RuntimeException types here
// Need to unwrap to ascertain real causing exception before we choose to ignore
// Ignore err-logging all transport-level/type exceptions、
// 在这里我们收到RuntimeException类型的异常，我们需要在忽略这个异常前拆开这个异常，查明弄清是什么异常，、
if (!isIgnorableException(x)) {
// Log the exception at error level and continue
LOGGER.error((x instanceof TException ? "Thrift " : "") + "Error occurred during processing of message.", x);
}
} finally {
if (eventHandler != null) {
eventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
}
if (inputTransport != null) {
inputTransport.close();
}
if (outputTransport != null) {
outputTransport.close();
}
if (client_.isOpen()) {
client_.close();
}
}
}
// 分析异常，看看是什么导致的异常，对Thrift的异常还不太了解，了解了以后在来研究下这
private boolean isIgnorableException(Exception x) {
TTransportException tTransportException = null;
if (x instanceof TTransportException) {
tTransportException = (TTransportException) x;
} else if (x.getCause() instanceof TTransportException) {
tTransportException = (TTransportException) x.getCause();
}
if (tTransportException != null) {
switch (tTransportException.getType()) {
case TTransportException.END_OF_FILE:
case TTransportException.TIMED_OUT:
return true;
}
}
return false;
}
}
网友评论