Tomcat使用线程池处理远程并发请求的方法
通过了解学习tomcat如何处理并发请求，了解到线程池，锁，队列，unsafe类，下面的主要代码来自
java-jre：
sun.misc.Unsafe java.util.concurrent.ThreadPoolExecutor java.util.concurrent.ThreadPoolExecutor.Worker java.util.concurrent.locks.AbstractQueuedSynchronizer java.util.concurrent.locks.AbstractQueuedLongSynchronizer java.util.concurrent.LinkedBlockingQueue
tomcat:
org.apache.tomcat.util.net.NioEndpoint org.apache.tomcat.util.threads.ThreadPoolExecutor org.apache.tomcat.util.threads.TaskThreadFactory org.apache.tomcat.util.threads.TaskQueue
ThreadPoolExecutor
是一个线程池实现类，管理线程，减少线程开销，可以用来提高任务执行效率，
构造方法中的参数有
publicThreadPoolExecutor( intcorePoolSize, intmaximumPoolSize, longkeepAliveTime, TimeUnitunit, BlockingQueueworkQueue, ThreadFactorythreadFactory, RejectedExecutionHandlerhandler){ }
corePoolSize是核心线程数
maximumPoolSize是最大线程数
keepAliveTime非核心线程最大空闲时间（超过时间终止）
unit时间单位
workQueue队列，当任务过多时，先存放在队列
threadFactory线程工厂，创建线程的工厂
handler决绝策略，当任务数过多，队列不能再存放任务时，该如何处理，由此对象去处理。这是个接口，你可以自定义处理方式
ThreadPoolExecutor在Tomcat中http请求的应用
此线程池是tomcat用来在接收到远程请求后，将每次请求单独作为一个任务去处理，每次调用execute(Runnable)
初始化
org.apache.tomcat.util.net.NioEndpoint
NioEndpoint初始化的时候，创建了线程池
publicvoidcreateExecutor(){
internalExecutor=true;
TaskQueuetaskqueue=newTaskQueue();
//TaskQueue无界队列，可以一直添加，因此handler等同于无效
TaskThreadFactorytf=newTaskThreadFactory(getName()+"-exec-",daemon,getThreadPriority());
executor=newThreadPoolExecutor(getMinSpareThreads(),getMaxThreads(),60,TimeUnit.SECONDS,taskqueue,tf);
taskqueue.setParent((ThreadPoolExecutor)executor);
}
在线程池创建时，调用prestartAllCoreThreads(),初始化核心工作线程worker，并启动
publicintprestartAllCoreThreads(){
intn=0;
while(addWorker(null,true))
++n;
returnn;
}
当addWorker数量等于corePoolSize时，addWorker(null,ture)会返回false,停止worker工作线程的创建
提交任务到队列
每次客户端过来请求（http），就会提交一次处理任务，
worker从队列中获取任务运行，下面是任务放入队列的逻辑代码
ThreadPoolExecutor.execute(Runnable)提交任务：
publicvoidexecute(Runnablecommand){
if(command==null)
thrownewNullPointerException();
intc=ctl.get();
//worker数是否小于核心线程数tomcat中初始化后，一般不满足第一个条件，不会addWorker
if(workerCountOf(c)
workQueue.offer(command)完成了任务的提交(在tomcat处理远程http请求时)。
workQueue.offer
TaskQueue是BlockingQueue具体实现类，workQueue.offer(command)实际代码：
publicbooleanoffer(Ee){
if(e==null)thrownewNullPointerException();
finalAtomicIntegercount=this.count;
if(count.get()==capacity)
returnfalse;
intc=-1;
Nodenode=newNode(e);
finalReentrantLockputLock=this.putLock;
putLock.lock();
try{
if(count.get()=0;
}
//添加任务到队列
/**
*Linksnodeatendofqueue.
*
*@paramnodethenode
*/
privatevoidenqueue(Nodenode){
//assertputLock.isHeldByCurrentThread();
//assertlast.next==null;
last=last.next=node;//链表结构last.next=node;last=node
}
之后是worker的工作，worker在run方法中通过去getTask()获取此处提交的任务，并执行完成任务。
线程池如何处理新提交的任务
添加worker之后，提交任务，因为worker数量达到corePoolSize，任务都会将放入队列，而worker的run方法则是循环获取队列中的任务（不为空时），
workerrun方法：
/**DelegatesmainrunlooptoouterrunWorker*/
publicvoidrun(){
runWorker(this);
}
循环获取队列中的任务
runWorker(worker)方法循环部分代码：
finalvoidrunWorker(Workerw){
Threadwt=Thread.currentThread();
Runnabletask=w.firstTask;
w.firstTask=null;
w.unlock();//allowinterrupts
booleancompletedAbruptly=true;
try{
while(task!=null||(task=getTask())!=null){//循环获取队列中的任务
w.lock();//上锁
try{
//运行前处理
beforeExecute(wt,task);
//队列中的任务开始执行
task.run();
//运行后处理
afterExecute(task,thrown);
}finally{
task=null;
w.completedTasks++;
w.unlock();//释放锁
}
}
completedAbruptly=false;
}finally{
processWorkerExit(w,completedAbruptly);
}
}
task.run()执行任务
锁运用
ThreadPoolExecutor使用锁主要保证两件事情，
1.给队列添加任务，保证其他线程不能操作队列
2.获取队列的任务，保证其他线程不能同时操作队列
给队列添加任务上锁
publicbooleanoffer(Ee){
if(e==null)thrownewNullPointerException();
finalAtomicIntegercount=this.count;
if(count.get()==capacity)
returnfalse;
intc=-1;
Nodenode=newNode(e);
finalReentrantLockputLock=this.putLock;
putLock.lock();//上锁
try{
if(count.get()=0;
}
获取队列任务上锁
privateRunnablegetTask(){
booleantimedOut=false;//Didthelastpoll()timeout?
//...省略
for(;;){
try{
Runnabler=timed?
workQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS):
workQueue.take();//获取队列中一个任务
if(r!=null)
returnr;
timedOut=true;
}catch(InterruptedExceptionretry){
timedOut=false;
}
}
}
publicEtake()throwsInterruptedException{
Ex;
intc=-1;
finalAtomicIntegercount=this.count;
finalReentrantLocktakeLock=this.takeLock;
takeLock.lockInterruptibly();//上锁
try{
while(count.get()==0){
notEmpty.await();//如果队列中没有任务，等待
}
x=dequeue();
c=count.getAndDecrement();
if(c>1)
notEmpty.signal();
}finally{
takeLock.unlock();//释放锁
}
if(c==capacity)
signalNotFull();
returnx;
}
volatile
在并发场景这个关键字修饰成员变量很常见，
主要目的公共变量在被某一个线程修改时，对其他线程可见（实时）
sun.misc.Unsafe高并发相关类
线程池使用中，有平凡用到Unsafe类，这个类在高并发中，能做一些原子CAS操作，锁线程，释放线程等。
sun.misc.Unsafe类是底层类，openjdk源码中有
原子操作数据
java.util.concurrent.locks.AbstractQueuedSynchronizer类中就有保证原子操作的代码
protectedfinalbooleancompareAndSetState(intexpect,intupdate){
//Seebelowforintrinsicssetuptosupportthis
returnunsafe.compareAndSwapInt(this,stateOffset,expect,update);
}
对应Unsafe类的代码:
//对应的java底层，实际是native方法，对应C++代码
/**
*AtomicallyupdateJavavariabletoxifitiscurrently
*holdingexpected.
*@returntrueifsuccessful
*/
publicfinalnativebooleancompareAndSwapInt(Objecto,longoffset,
intexpected,
intx);
方法的作用简单来说就是更新一个值，保证原子性操作
当你要操作一个对象o的一个成员变量offset时,修改o.offset，
高并发下为保证准确性，你在操作o.offset的时候，读应该是正确的值，并且中间不能被别的线程修改来保证高并发的环境数据操作有效。
即expected期望值与内存中的值比较是一样的expected==内存中的值，则更新值为x，返回true代表修改成功
否则，期望值与内存值不同，说明值被其他线程修改过，不能更新值为x，并返回false，告诉操作者此次原子性修改失败。
阻塞和唤醒线程
publicnativevoidpark(booleanisAbsolute,longtime);//阻塞当前线程
线程池的worker角色循环获取队列任务，如果队列中没有任务，worker.run还是在等待的，不会退出线程，代码中用了notEmpty.await()中断此worker线程，放入一个等待线程队列（区别去任务队列）；当有新任务需要时，再notEmpty.signal()唤醒此线程
底层分别是
unsafe.park()阻塞当前线程
publicnativevoidpark(booleanisAbsolute,longtime);
unsafe.unpark()唤醒线程
publicnativevoidunpark(Objectthread);
这个操作是对应的，阻塞时，先将thread放入队列，唤醒时，从队列拿出被阻塞的线程，unsafe.unpark(thread)唤醒指定线程。
java.util.concurrent.locks.AbstractQueuedLongSynchronizer.ConditionObject类中
通过链表存放线程信息
//添加一个阻塞线程
privateNodeaddConditionWaiter(){
Nodet=lastWaiter;
//IflastWaiteriscancelled,cleanout.
if(t!=null&&t.waitStatus!=Node.CONDITION){
unlinkCancelledWaiters();
t=lastWaiter;
}
Nodenode=newNode(Thread.currentThread(),Node.CONDITION);
if(t==null)
firstWaiter=node;
else
t.nextWaiter=node;
lastWaiter=node;//将新阻塞的线程放到链表尾部
returnnode;
}
//拿出一个被阻塞的线程
publicfinalvoidsignal(){
if(!isHeldExclusively())
thrownewIllegalMonitorStateException();
Nodefirst=firstWaiter;//链表中第一个阻塞的线程
if(first!=null)
doSignal(first);
}
//拿到后，唤醒此线程
finalbooleantransferForSignal(Nodenode){
LockSupport.unpark(node.thread);
returntrue;
}
publicstaticvoidunpark(Threadthread){
if(thread!=null)
UNSAFE.unpark(thread);
}
到此这篇关于Tomcat使用线程池处理远程并发请求的方法的文章就介绍到这了,更多相关Tomcat线程池处理远程并发请求内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！