服务容错之限流之 Tomcat 限流 Tomcat 线程池的拒绝策略

在文章开头，先和大家抛出两个问题：

  1. 每次提到服务限流为什么都不考虑基于 Tomcat 来做呢？
  2. 大家有遇到过 Tomcat 线程池触发了拒绝策略吗？

JUC 线程池

在谈 Tomcat 的线程池前，先看一下 JUC 中线程池的执行流程，这里使用《Java 并发编程的艺术》中的一张图：
在这里插入图片描述

即执行流程为：

  1. 收到提交任务
  2. 当前线程数小于核心线程数，创建一个新的线程来执行任务
  3. 当前线程数大于等于核心线程数，
    • 如果阻塞队列未满，将任务存储到队列
    • 如果阻塞队列已满
      • 如果当前线程数小于最大线程数，则创建一个线程来执行新提交的任务
      • 如果当前线程数大于等于最大线程数，执行拒绝策略

可以看到设计思想是任务可以等待执行，但要尽量少的创造过多线程。如果队列很大，则很难扩大到最大线程数，同时会有大量的任务等待。

Tomcat 线程池分析

Tomcat 线程池是在 LifeCycle 中创建的。跳过前面繁琐的流程，直接看 org.apache.tomcat.util.net.NioEndpoint#startInternal：

    /*** Start the NIO endpoint, creating acceptor, poller threads.*/@Overridepublic void startInternal() throws Exception {if (!running) {running = true;paused = false;processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,socketProperties.getProcessorCache());eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,socketProperties.getEventCache());nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,socketProperties.getBufferPool());// Create worker collectionif ( getExecutor() == null ) {createExecutor();}initializeConnectionLatch();// Start poller threadspollers = new Poller[getPollerThreadCount()];for (int i=0; i<pollers.length; i++) {pollers[i] = new Poller();Thread pollerThread = new Thread(pollers[i], getName() + "-ClientPoller-"+i);pollerThread.setPriority(threadPriority);pollerThread.setDaemon(true);pollerThread.start();}startAcceptorThreads();}}

再看 org.apache.tomcat.util.net.AbstractEndpoint#createExecutor：

    public void createExecutor() {internalExecutor = true;TaskQueue taskqueue = new TaskQueue(); //无界队列TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority());executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf);taskqueue.setParent( (ThreadPoolExecutor) executor);}

要注意这里的 ThreadPoolExecutor 不是 JUC 里面的 java.util.concurrent.ThreadPoolExecutor，而是 Tomcat 的 org.apache.tomcat.util.threads.ThreadPoolExecutor，它继承了 JUC 的 java.util.concurrent.ThreadPoolExecutor：

public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {...
}

查看它的构造方法：

    public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new RejectHandler());//提前启动核心线程prestartAllCoreThreads();}

可以发现它在构造的时候就会启动核心线程，而 java.util.concurrent.ThreadPoolExecutor 则是需要手动启动。而阻塞队列使用是 org.apache.tomcat.util.threads.TaskQueue：

public class TaskQueue extends LinkedBlockingQueue<Runnable> {private static final long serialVersionUID = 1L;private volatile ThreadPoolExecutor parent = null;// No need to be volatile. This is written and read in a single thread// (when stopping a context and firing the  listeners)private Integer forcedRemainingCapacity = null;public TaskQueue() {super();}...
}

而在创建 org.apache.tomcat.util.threads.TaskQueue 的时候，并没有传递 capacity，也就是说 Tomcat 的线程池使用的是无界队列。

接下来看一下最核心的org.apache.tomcat.util.threads.ThreadPoolExecutor#execute(java.lang.Runnable)：

/*** {@inheritDoc}*/@Overridepublic void execute(Runnable command) {//重载 java.util.concurrent.ThreadPoolExecutor#executeexecute(command,0,TimeUnit.MILLISECONDS);}public void execute(Runnable command, long timeout, TimeUnit unit) {submittedCount.incrementAndGet();try {super.execute(command);} catch (RejectedExecutionException rx) {if (super.getQueue() instanceof TaskQueue) {final TaskQueue queue = (TaskQueue)super.getQueue();try {if (!queue.force(command, timeout, unit)) {submittedCount.decrementAndGet();throw new RejectedExecutionException("Queue capacity is full.");}} catch (InterruptedException x) {submittedCount.decrementAndGet();throw new RejectedExecutionException(x);}} else {submittedCount.decrementAndGet();throw rx;}}}

本质上还是执行的 java.util.concurrent.ThreadPoolExecutor#execute 方法：

    public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();// Tomcat 中这块逻辑不会执行，因为构造时已经初始化了核心线程if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}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);}//强制入队public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if ( parent==null || parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected}

这里的 workQueue 是 org.apache.tomcat.util.threads.TaskQueue， org.apache.tomcat.util.threads.TaskQueue#offer：

    @Overridepublic boolean offer(Runnable o) {//we can't do any checksif (parent==null) return super.offer(o);//we are maxed out on threads, simply queue the object//当前线程数达到最大，任务入队if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);//we have idle threads, just add it to the queue//如果已提交未执行完的任务数小于当前线程数（来了任务先+1，再入队，执行完才-1，说明还有空闲的worker线程）,任务入队if (parent.getSubmittedCount()<(parent.getPoolSize())) return super.offer(o);//if we have less threads than maximum force creation of a new thread// 如果当前线程数小于最大线程数量，则直接返回false，java.util.concurrent.ThreadPoolExecutor#execute 会创建新的线程来执行任务if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;//if we reached here, we need to add it to the queue//任务入队（当前线程数大于最大线程数）return super.offer(o);}

再看下拒绝策略，结合 java.util.concurrent.ThreadPoolExecutor#execute 方法，需要 java.util.concurrent.ThreadPoolExecutor#addWorker 返回 false 才会触发，即达到了最大线程数才会触发，而 org.apache.tomcat.util.threads.ThreadPoolExecutor#execute(java.lang.Runnable) 在触发了拒绝策略后还有一个特殊处理：

					//如果是 TaskQueueif (super.getQueue() instanceof TaskQueue) {final TaskQueue queue = (TaskQueue)super.getQueue();try {//强制入队if (!queue.force(command, timeout, unit)) {submittedCount.decrementAndGet();throw new RejectedExecutionException("Queue capacity is full.");}} catch (InterruptedException x) {submittedCount.decrementAndGet();throw new RejectedExecutionException(x);}} else { //非 TaskQueue 直接触发拒绝策略submittedCount.decrementAndGet();throw rx;}

再看 org.apache.tomcat.util.threads.TaskQueue#force(java.lang.Runnable, long, java.util.concurrent.TimeUnit)：

    public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if ( parent==null || parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected}

说白了就是直接入队（无界队列）：

    public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {if (e == null) throw new NullPointerException();long nanos = unit.toNanos(timeout);int c = -1;final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;putLock.lockInterruptibly();try {while (count.get() == capacity) { //capacity是Integer最大值if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}enqueue(new Node<E>(e));c = count.getAndIncrement();if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}if (c == 0)signalNotEmpty();return true;}

这么看，Tomcat 的线程池基本上不会触发拒绝策略。可以写个例子试一下：

package blog.dongguabai.others.tomcat_threadpool;import org.apache.tomcat.util.threads.TaskQueue;
import org.apache.tomcat.util.threads.TaskThreadFactory;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;import java.util.Date;
import java.util.concurrent.TimeUnit;/*** @author dongguabai* @date 2023-11-18 22:04*/
public class Demo {public static void main(String[] args) {//无界队列TaskQueue taskqueue = new TaskQueue();TaskThreadFactory tf = new TaskThreadFactory("dongguabai_blog" + "-exec-", false, 2);final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 60, TimeUnit.SECONDS, taskqueue, tf);taskqueue.setParent(executor);observe(executor);while (true) {executor.execute(new Runnable() {public void run() {excuteForever();}});}}private static void observe(final ThreadPoolExecutor executor) {Runnable task = new Runnable() {public void run() {while (true) {try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(new Date().toLocaleString() + "->" + executor.getQueue().size());}}};new Thread(task).start();}public static void excuteForever() {while (true) {}}
}

输出：

2023-11-18 22:18:27->6541506
2023-11-18 22:18:34->14395417
2023-11-18 22:18:37->25708908
2023-11-18 22:18:50->32014458
2023-11-18 22:19:07->47236736
2023-11-18 22:19:10->65616058
2023-11-18 22:19:32->66856933
...

可以看到，队列里的任务都有六千多万了，还没有触发拒绝策略，线程池还是可以继续接收任务。

当然我们也是可以自定义的，只需要重写 org.apache.tomcat.util.net.AbstractEndpoint#getExecutor 即可：

    public Executor getExecutor() { return executor; }

org.apache.tomcat.util.net.NioEndpoint#startInternal 会进行判断：

@Overridepublic void startInternal() throws Exception {if (!running) {...if ( getExecutor() == null ) {createExecutor(); //如果没有自定义实现，就会使用默认实现}}...}

Tomcat 默认线程池优先创建线程执行任务，达到了最大线程数，不会直接执行拒绝策略，而是尝试返回等待队列，但由于等待队列的容量是 Integer 最大值，所以几乎不会触发拒绝策略。

最后

最后再回过头看文章开头的两个问题：

  1. 每次提到服务限流为什么都不考虑基于 Tomcat 来做呢？

    Tomcat 的确可以用来做限流，比如可以控制最大线程数，这样后续的任务均会在队列等待，并不会执行。org.apache.tomcat.util.net.AbstractEndpoint#setMaxConnections 从 Connector 的角度设置，这块不在本文探讨范围之内。

    虽然基于 Tomcat 的限流是一种可能的方案，但在实际应用中，我们通常会选择其他的层次来实现服务限流：

    • 可扩展性：基于 Tomcat 的限流方案通常只能在单个服务实例上工作，且只能针对HTTP/HTTPS协议的请。而在微服务或者分布式系统中，我们可能需要分布式限流方案和针对多协议的 限流。
    • 灵活性：在应用层或者分布式系统层实现的限流方案通常可以提供更多的配置选项和更精细的控制。例如，请求的资源、来源或者其他属性来进行限流。

  2. 大家有遇到过 Tomcat 线程池触发了拒绝策略吗？

    Tomcat 默认无限队列，难以触发拒绝策略，所以会有内存泄漏的风险。可以基于 org.apache.tomcat.util.net.AbstractEndpoint#getExecutor 自定义线程池进行控制。

References

  • 《Java 并发编程的艺术》

欢迎关注公众号：
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/147114.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Java中利用OpenCV进行人脸识别

Java中利用OpenCV进行人脸识别

OpenCV 概述 ​ OpenCV&#xff08;Open Source Computer Vision Library&#xff09;是一个开源计算机视觉库&#xff0c;它提供了丰富的工具和算法&#xff0c;用于处理图像和视频数据。该库由一系列高效的计算机视觉算法组成&#xff0c;涵盖了许多领域&#xff0c;包括目…
阅读更多...
近几天接触的自动化框架，支持Android、Web和Windows

近几天接触的自动化框架，支持Android、Web和Windows

工作主要做web&#xff0c; 简单了解过selenium自动化框架&#xff0c;使用的不深&#xff0c; 但在工作中确实帮我节省了好多时间。 主要应用于数据的导入导出。 最近几天开始了解Aibote 。 主要还是了解web自动化。这里先做记录&#xff0c;后面看看是不是可以做个例子出来。…
阅读更多...
修改服务器端Apache默认根目录

修改服务器端Apache默认根目录

目标&#xff1a;修改默认Apache网站根目录 /var/www/html 一、找到 DocumentRoot “/var/www/html” 这一段 apache的根目录&#xff0c;把/var/www/html 这个目录改 #DocumentRoot "/var/www/html" DocumentRoot "/home/cloud/tuya_mini_h5/build" 二、…
阅读更多...
小程序如何添加打印机来打印订单

小程序如何添加打印机来打印订单

在采云小程序中&#xff0c;支持打印订单的小票、标签、发货单和电子面单。小票打印机用于打印小票&#xff0c;类似于超市小票、外卖小票等。标签打印机用于打印商品标签&#xff0c;类似于奶茶上面粘贴的标签&#xff0c;用于表示饮料名称和规格等。货单打印机用于打印发货单…
阅读更多...
【好奇心驱动力】ESP8266驱动SG90舵机开关灯

【好奇心驱动力】ESP8266驱动SG90舵机开关灯

0.前言 ESP8266弄丢了好几个都忘记放在哪&#xff0c;重新买了个typeC接口的方便多了&#xff0c;看到驱动SG90舵机作为智能开关&#xff0c;简单复现了一下&#xff0c;代码比较简单&#xff0c;没有连接小爱同学或者其他语音助手。 1.实验方法 ESP8266连接SG90舵机&#x…
阅读更多...
32位单片机PY32F040，主频72M，外设丰富，支持断码LCD

32位单片机PY32F040，主频72M，外设丰富，支持断码LCD

PY32F040 系列微控制器采用高性能的 32 位 ARM Cortex-M0 内核,宽电压工作范围的 MCU。嵌入高达 128 Kbytes flash 和 16 Kbytes SRAM 存储器,最高工作频率 72 MHz。LQFP64封装两块出头就可以拿到&#xff0c;我们还有开发板和开发资料帮助客户更好的开发。 PY32F040 系列微控…
阅读更多...
C# NAudio 音频库

C# NAudio 音频库

C# NAudio 音频库 NAudio安装NAudio简述简单示例1录制麦克风录制系统声卡WAV格式播放MP3格式播放AudioFileReader读取播放音频MediaFoundationReader 读取播放音频 NAudio安装 项目>NuGet包管理器 搜索NAudio点击安装&#xff0c;自动安装依赖库。 安装成功后工具箱会新增…
阅读更多...
视频推拉流EasyDSS直播点播平台获取指定时间快照的实现方法

视频推拉流EasyDSS直播点播平台获取指定时间快照的实现方法

视频推拉流直播点播系统EasyDSS平台&#xff0c;可提供流畅的视频直播、点播、视频推拉流、转码、管理、分发、录像、检索、时移回看等功能&#xff0c;可兼容多操作系统&#xff0c;在直播点播领域具有广泛的场景应用。为了便于用户集成、调用与二次开发。 今天我们来介绍下在…
阅读更多...
WPF实现右键菜单

WPF实现右键菜单

在WPF中&#xff0c;创建上下文菜单&#xff08;通常称为“右键菜单”&#xff09;是通过使用ContextMenu控件来实现的。你可以在XAML中声明上下文菜单&#xff0c;并将其关联到任何FrameworkElement。以下是如何在WPF中实现上下文菜单的基本步骤&#xff1a; 1. 在XAML中定义…
阅读更多...
KT148A语音芯片使用串口uart本控制的完整说明_包含硬件和指令举例

KT148A语音芯片使用串口uart本控制的完整说明_包含硬件和指令举例

一、功能简介 KT148A肯定是支持串口的&#xff0c;有客户反馈使用一线还是不方便&#xff0c;比如一些大型的系统不适合有延时的操作&#xff0c;所以更加倾向于使用uart控制&#xff0c;这里我们也给出解决方案 延伸出来另外一个版本&#xff0c;KT158A 注意次版本芯片还是…
阅读更多...
gRPC 四模式之 一元RPC模式

gRPC 四模式之 一元RPC模式

一元RPC模式 一元 RPC 模式也被称为简单 RPC 模式。在该模式中&#xff0c;当客户端调用服务器端的远程方法时&#xff0c;客户端发送请求至服务器端并获得一个响应&#xff0c;与响应一起发送的还有状态细节以及 trailer 元数据&#xff08;这部分不是默认发送的&#xff0c;…
阅读更多...
.L0CK3D来袭：如何保护您的数据免受致命攻击

.L0CK3D来袭：如何保护您的数据免受致命攻击

尊敬的读者&#xff1a; 网络犯罪的威胁日益增长&#xff0c;其中.L0CK3D勒索病毒是一种极具挑战性的数字威胁。为了助您应对这一风险&#xff0c;本文将深入探讨.L0CK3D病毒的狡猾手法、毁灭性影响&#xff0c;提供详实的数据恢复方法&#xff0c;同时为您提供极具实战性的预…
阅读更多...
流量2----2

流量2----2

2
阅读更多...
ArcGIS Maps SDK for JS：监听图层的visible属性

ArcGIS Maps SDK for JS：监听图层的visible属性

文章目录 1 问题描述2 解决方案3 拓展 1 问题描述 近期有这么一个需求。在 ArcGIS Maps SDK for JavaScript 中&#xff0c;使用图层的visible属性同步显示某个组件&#xff0c;即打开图层时显示组件&#xff0c;关闭图层时隐藏组件。 首先想到的是&#xff0c;通过点击图层列…
阅读更多...
HTTP 到 HTTPS 再到 HSTS 的转变

HTTP 到 HTTPS 再到 HSTS 的转变

近些年&#xff0c;随着域名劫持、信息泄漏等网络安全事件的频繁发生&#xff0c;网站安全也变得越来越重要&#xff0c;也促成了网络传输协议从 HTTP 到 HTTPS 再到 HSTS 的转变。 HTTP HTTP&#xff08;超文本传输协议&#xff09; 是一种用于分布式、协作式和超媒体信息系…
阅读更多...
【总结】各层的校验和的特点

【总结】各层的校验和的特点

数据链路层 MAC帧头的校验码&#xff08;FCS&#xff09;&#xff08;4B&#xff09;&#xff0c;校验首部和数据&#xff0c;即从目的地址到数据尾的所有字节&#xff08;共有数据 64B-1518B&#xff09;&#xff08;4618-150018&#xff09; 网络层 IP帧头的首部校验和&am…
阅读更多...
OpenAI的多函数调用（Multiple Function Calling)简介

OpenAI的多函数调用（Multiple Function Calling)简介

我在六月份写了一篇关于GPT 函数调用&#xff08;Function calling) 的博客https://blog.csdn.net/xindoo/article/details/131262670&#xff0c;其中介绍了函数调用的方法&#xff0c;但之前的函数调用&#xff0c;在一轮对话中只能调用一个函数。就在上周&#xff0c;OpenAI…
阅读更多...
C# IEnumerable＜T＞介绍

C# IEnumerable＜T＞介绍

IEnumerable 是 C# 中的一个接口&#xff0c;它是 .NET Framework 中的集合类型的基础。任何实现了 IEnumerable 接口的对象都可以进行 foreach 迭代。 IEnumerable 只有一个方法&#xff0c;即 GetEnumerator&#xff0c;该方法返回一个 IEnumerator 对象。IEnumerator 对象用…
阅读更多...
Ubuntu22.04 部署Mqtt服务器

Ubuntu22.04 部署Mqtt服务器

1、打开Download EMQX &#xff08;www.emqx.io&#xff09;下载mqtt服务器版本 2、Download the EMQX repository curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash 3.Install EMQX sudo apt-get install emqx 4.Run EMQX sudo systemctl start…
阅读更多...
开源与闭源：数字化时代的辩论与未来走向

开源与闭源：数字化时代的辩论与未来走向

在当今的数字化时代&#xff0c;关于开源和闭源软件的辩论一直是技术界的热门话题。 特斯拉CEO马斯克最近也加入了这场辩论&#xff0c;公开表示OpenAI不应该闭源&#xff0c;而他自己的首款聊天机器人将选择开源。 这引发了人们对开源与闭源软件的进一步思考&#xff1a;开源是…
阅读更多...
最新文章

Copyright @ 2022~2023