在使用和配置tomcat的线程池之前明白线程池的原理，类似于操作系统中的缓冲区的概念，它的流程如下：先启动若干数量的线程，并让这些线程都处于睡眠 状态，当客户端有一个新请求时，就会唤醒线程池中的某一个睡眠线程，让它来处理客户端的这个请求，当处理完这个请求后，线程又处于睡眠状态。可能你也许会 问：为什么要搞得这么麻烦，如果每当客户端有新的请求时，我就创建一个新的线程不就完了？这也许是个不错的方法，因为它能使得你编写代码相对容易一些，但 你却忽略了一个重要的问题??性能！
通过了解学习tomcat如何处理并发请求，了解到线程池，锁，队列，unsafe类，下面的主要代码来自
java-jre：
1 2 3 4 5 6 | 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:
1 2 3 4 | 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
是一个线程池实现类，管理线程，减少线程开销，可以用来提高任务执行效率，
构造方法中的参数有
1 2 3 4 5 6 7 8 9 10 | public ThreadPoolExecutor( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { } |
corePoolSize 是核心线程数
maximumPoolSize 是最大线程数
keepAliveTime 非核心线程最大空闲时间（超过时间终止）
unit 时间单位
workQueue 队列，当任务过多时，先存放在队列
threadFactory 线程工厂，创建线程的工厂
handler 决绝策略，当任务数过多，队列不能再存放任务时，该如何处理，由此对象去处理。这是个接口，你可以自定义处理方式
ThreadPoolExecutor在Tomcat中http请求的应用
此线程池是tomcat用来在接收到远程请求后，将每次请求单独作为一个任务去处理，每次调用execute(Runnable)
初始化
org.apache.tomcat.util.net.NioEndpoint
NioEndpoint初始化的时候，创建了线程池
1 2 3 4 5 6 7 8 | public void createExecutor() { internalExecutor = true; TaskQueue taskqueue = new TaskQueue(); //TaskQueue无界队列，可以一直添加，因此handler 等同于无效 TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority()); executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf); taskqueue.setParent( (ThreadPoolExecutor) executor); } |
在线程池创建时，调用prestartAllCoreThreads(), 初始化核心工作线程worker，并启动
1 2 3 4 5 6 | public int prestartAllCoreThreads() { int n = 0; while (addWorker(null, true)) ++n; return n; } |
当addWorker 数量等于corePoolSize时，addWorker(null,ture)会返回false,停止worker工作线程的创建
提交任务到队列
每次客户端过来请求（http），就会提交一次处理任务，
worker 从队列中获取任务运行，下面是任务放入队列的逻辑代码
ThreadPoolExecutor.execute(Runnable) 提交任务：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 1 | public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // worker数 是否小于 核心线程数 tomcat中初始化后，一般不满足第一个条件，不会addWorker if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // workQueue.offer(command)，将任务添加到队列， if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); } |
workQueue.offer(command) 完成了任务的提交(在tomcat处理远程http请求时)。
workQueue.offer
TaskQueue 是 BlockingQueue 具体实现类，workQueue.offer(command)实际代码：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 35 | public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; if (count.get() == capacity) return false; int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; putLock.lock(); try { if (count.get() < capacity) { enqueue(node); //此处将任务添加到队列 c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); return c >= 0; } // 添加任务到队列 /** * Links node at end of queue. * * @param node the node */ private void enqueue(Node<E> node) { // assert putLock.isHeldByCurrentThread(); // assert last.next == null; last = last.next = node; //链表结构 last.next = node; last = node |
之后是worker的工作，worker在run方法中通过去getTask()获取此处提交的任务，并执行完成任务。
线程池如何处理新提交的任务
添加worker之后，提交任务，因为worker数量达到corePoolSize，任务都会将放入队列，而worker的run方法则是循环获取队列中的任务（不为空时），
worker run方法：
1 2 3 4 | /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); } |
循环获取队列中的任务
runWorker(worker)方法 循环部分代码：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = 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.获取队列的任务，保证其他线程不能同时操作队列
给队列添加任务上锁
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 | public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; if (count.get() == capacity) return false; int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; putLock.lock(); //上锁 try { if (count.get() < capacity) { enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } } finally { putLock.unlock(); //释放锁 } if (c == 0) signalNotEmpty(); return c >= 0; } |
获取队列任务上锁
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 | private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? // ...省略 for (;;) { try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); //获取队列中一个任务 if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = 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(); return x; } |
volatile
在并发场景这个关键字修饰成员变量很常见，
主要目的公共变量在被某一个线程修改时，对其他线程可见（实时）
sun.misc.Unsafe 高并发相关类
线程池使用中，有平凡用到Unsafe类，这个类在高并发中，能做一些原子CAS操作，锁线程，释放线程等。
sun.misc.Unsafe 类是底层类，openjdk源码中有
原子操作数据
java.util.concurrent.locks.AbstractQueuedSynchronizer 类中就有保证原子操作的代码
1 2 3 4 | protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); } |
对应Unsafe类的代码:
1 2 3 4 5 6 7 8 9 | //对应的java底层，实际是native方法，对应C++代码 /** * Atomically update Java variable to <tt>x</tt> if it is currently * holding <tt>expected</tt>. * @return <tt>true</tt> if successful */ public final native boolean compareAndSwapInt(Object o, long offset, int expected, int x); |
方法的作用简单来说就是 更新一个值，保证原子性操作
当你要操作一个对象o的一个成员变量offset时,修改o.offset，
高并发下为保证准确性，你在操作o.offset的时候，读应该是正确的值，并且中间不能被别的线程修改来保证高并发的环境数据操作有效。
即 expected 期望值与内存中的值比较是一样的expected == 内存中的值 ，则更新值为 x，返回true代表修改成功
否则，期望值与内存值不同，说明值被其他线程修改过，不能更新值为x，并返回false，告诉操作者此次原子性修改失败。
阻塞和唤醒线程
public native void park(boolean isAbsolute, long time); //阻塞当前线程
线程池的worker角色循环获取队列任务，如果队列中没有任务，worker.run 还是在等待的，不会退出线程，代码中用了notEmpty.await() 中断此worker线程，放入一个等待线程队列（区别去任务队列）；当有新任务需要时，再notEmpty.signal()唤醒此线程
底层分别是
unsafe.park() 阻塞当前线程
public native void park(boolean isAbsolute, long time);
unsafe.unpark() 唤醒线程
public native void unpark(Object thread);
这个操作是对应的，阻塞时，先将thread放入队列，唤醒时，从队列拿出被阻塞的线程，unsafe.unpark(thread)唤醒指定线程。
java.util.concurrent.locks.AbstractQueuedLongSynchronizer.ConditionObject 类中
通过链表存放线程信息
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | // 添加一个阻塞线程 private Node addConditionWaiter() { Node t = lastWaiter; // If lastWaiter is cancelled, clean out. if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) firstWaiter = node; else t.nextWaiter = node; lastWaiter = node; //将新阻塞的线程放到链表尾部 return node; } // 拿出一个被阻塞的线程 public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; //链表中第一个阻塞的线程 if (first != null) doSignal(first); } // 拿到后，唤醒此线程 final boolean transferForSignal(Node node) { LockSupport.unpark(node.thread); return true; } public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); } |
用线程池，用较少的线程处理较多的访问，可以提高tomcat处理请求的能力。
1：配置executor属性
打开/conf/server.xml文件，在Connector之前配置一个线程池（这个executor可以自己手动去掉注释）：
<Executor name="myThreadPool"
namePrefix="catalina-exec-"
maxThreads="250"
maxIdleTime="60000"
prestartminSpareThreads="true"
minSpareThreads="50"/>
重要参数说明：
name：共享线程池的名字。这是Connector为了共享线程池要引用的名字，该名字必须唯一。默认值：None；
namePrefix:在JVM上，每个运行线程都可以有一个name 字符串。这一属性为线程池中每个线程的name字符串设置了一个前缀，Tomcat将把线程号追加到这一前缀的后面。默认值：tomcat-exec-；
maxThreads：该线程池可以容纳的最大线程数。默认值：200，一般建议设置500~ 800 ，要根据自己的硬件设施条件和实际业务需求而定。
maxIdleTime：在Tomcat关闭一个空闲线程之前，允许空闲线程持续的时间(以毫秒为单位)。只有当前活跃的线程数大于minSpareThread的值，才会关闭空闲线程。默认值：60000(一分钟)。
minSpareThreads：Tomcat应该始终打开的最小不活跃线程数。默认值：25。Tomcat启动初始化的线程数，在tomcat初始化的时候就初始化
2：配置Connector
<Connector executor="myThreadPool"
port="8080" protocol="HTTP/1.1"
connectionTimeout="20000"
redirectPort="8443"
minProcessors="5"
maxProcessors="75"
acceptCount="1000"/>
重要参数说明：
executor：表示使用该参数值对应的线程池；
minProcessors：服务器启动时创建的处理请求的线程数；
maxProcessors：最大可以创建的处理请求的线程数；
acceptCount：指定当所有可以使用的处理请求的线程数都被使用时，可以放到处理队列中的请求数，超过这个数的请求将不予处理。