Tomcat使用线程池处理远程并发请求的方法

这篇文章主要介绍了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

是一个线程池实现类,管理线程,减少线程开销,可以用来提高任务执行效率,

构造方法中的参数有

 public ThreadPoolExecutor( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue 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初始化的时候,创建了线程池

 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,并启动

 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) 提交任务:

 public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // worker数 是否小于 核心线程数 tomcat中初始化后,一般不满足第一个条件,不会addWorker if (workerCountOf(c) 

workQueue.offer(command) 完成了任务的提交(在tomcat处理远程http请求时)。

workQueue.offer

TaskQueue 是 BlockingQueue 具体实现类,workQueue.offer(command)实际代码:

 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 node = new Node(e); final ReentrantLock putLock = this.putLock; putLock.lock(); try { if (count.get() = 0; } // 添加任务到队列 /** * Links node at end of queue. * * @param node the node */ private void enqueue(Node 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方法:

 /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); }

循环获取队列中的任务

runWorker(worker)方法 循环部分代码:

 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.获取队列的任务,保证其他线程不能同时操作队列

给队列添加任务上锁

 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 node = new Node(e); final ReentrantLock putLock = this.putLock; putLock.lock(); //上锁 try { if (count.get() = 0; }

 

获取队列任务上锁

 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 类中就有保证原子操作的代码

 protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }

对应Unsafe类的代码:

 //对应的java底层,实际是native方法,对应C++代码 /** * Atomically update Java variable to x if it is currently * holding expected. * @return true 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 类中

通过链表存放线程信息

 // 添加一个阻塞线程 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使用线程池处理远程并发请求的方法的详细内容,更多请关注0133技术站其它相关文章!

赞(0) 打赏
未经允许不得转载:0133技术站首页 » 服务器