当前位置: 首页 > news >正文

C++简易线程池

原理说明:

1. 线程池创建时,指定线程池的大小thread_size。当有新的函数任务通过函数addFunction ()添加进来后,其中一个线程执行函数。一个线程一次执行一个函数。如果函数数量大与线程池数量,则后来的函数等待。

2. 线程池内部有个容器m_functions 来存储待执行的函数。函数执行后从队列中移除。

3.  stopAll()函数会停止线程池。

ThreadPool.h

//ThreadPool.h
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>class ThreadPool {
public:static ThreadPool* getInstance(size_t thread_size = 1);		//默认线程池大小void addFunction(std::function<void()> task);				//添加需要执行的函数~ThreadPool();void stopAll(bool immediately);						//停止线程池, immediately:true立即停止, immediately:false等待当前线程函数执行完后停止。
private:ThreadPool(size_t thread_size);private:void workerThreadHandler();std::vector<std::thread> m_workers;					//线程容器std::queue<std::function<void()>> m_functions;			//待执行的函数容器std::mutex m_queue_mutex;std::condition_variable m_condition;bool m_stop;										//线程池停止状态std::thread m_wakeTimerThread;std::mutex m_timer_mutex;
};

ThreadPool.cpp

#include "ThreadPool.h"
#include <chrono>
#include <memory>
#include <thread>
#include <pthread.h>
#include <iostream>
using namespace std;ThreadPool* ThreadPool::getInstance(size_t thread_size)
{static std::mutex m_lock;static std::shared_ptr<ThreadPool> m_instance=nullptr;if (nullptr == m_instance){m_instance.reset(new ThreadPool(thread_size));}return m_instance.get();
}
ThreadPool::ThreadPool(size_t thread_size) : m_stop(false){m_workers.reserve(thread_size);for (size_t i=0; i<thread_size; ++i){m_workers.emplace_back([this, i](){	//创建线程池中的线程workerThreadHandler();});}//辅助线程,每隔一段时间发送一次唤醒,防止线程阻塞m_wakeTimerThread = std::thread([this](){	for(;!this->m_stop;){std::unique_lock<std::mutex> lock(this->m_timer_mutex);std::this_thread::sleep_for(std::chrono::milliseconds(2000));pthread_testcancel();m_condition.notify_all();std::cout<<"wake up"<<std::endl;}});std::cout<<__func__<<std::endl;
}//线程循环函数,循环查询函数容器是否为空,不为空则读取一个函数并执行。
void ThreadPool::workerThreadHandler()	
{for (;!this->m_stop;){std::function<void()> task;{std::unique_lock<std::mutex> lock(this->m_queue_mutex);std::cout<<"tasks begin size:"<<this->m_functions.size()<<" stop:"<<m_stop<<std::endl;if (!m_stop && m_functions.empty()){this->m_condition.wait(lock);}pthread_testcancel();	//作为线程的终止点if (this->m_stop){return;}if (this->m_functions.empty()){continue;}task = std::move(this->m_functions.front());this->m_functions.pop();std::cout<<std::this_thread::get_id() <<" tasks end size:"<<this->m_functions.size()<<std::endl;}task();std::cout<<__func__<<" end task"<<std::endl;std::this_thread::sleep_for(std::chrono::milliseconds(200));}
}void ThreadPool::addFunction(std::function<void()> task)	//添加待执行的函数
{std::unique_lock<std::mutex> lock(m_queue_mutex);if (m_stop){return;}m_functions.emplace(std::move(task));m_condition.notify_one();
}ThreadPool::~ThreadPool()
{{std::unique_lock<std::mutex> lock(m_queue_mutex);m_stop = true;}m_condition.notify_all();for (std::thread &worker: m_workers){worker.join();}m_wakeTimerThread.join();std::cout<<__func__<<std::endl;
}void ThreadPool::stopAll(bool immediately)	//停止线程, immediately:true立即停止
{this->m_stop = true;if (immediately){for (std::thread &worker: m_workers){pthread_cancel(worker.native_handle());}pthread_cancel(m_wakeTimerThread.native_handle());}
}

测试程序main.cpp

#include <iostream>
#include <chrono>
#include <mutex>
#include "ThreadPool.h"using namespace std;static std::mutex m_mutex;
void ProcessFunc111()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc222()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc333()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc444()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc555()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc666()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}int main()
{ThreadPool::getInstance(5);ThreadPool::getInstance()->addFunction([](){ProcessFunc111();});ThreadPool::getInstance()->addFunction([](){ProcessFunc222();});ThreadPool::getInstance()->addFunction([](){ProcessFunc333();});ThreadPool::getInstance()->addFunction([](){ProcessFunc444();});ThreadPool::getInstance()->addFunction([](){ProcessFunc555();});ThreadPool::getInstance()->addFunction([](){ProcessFunc666();});getchar();return 0;
}

执行结果:

tasks begin size:0 stop:0
ThreadPooltasks begin size:
0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
140563906656000 tasks end size:2
ProcessFunc111 begin
140563898263296 tasks end size:4
ProcessFunc222 begin
140563881477888 tasks end size:3
ProcessFunc333 begin
140563743504128 tasks end size:2
ProcessFunc444 begin
140563889870592 tasks end size:1
ProcessFunc555 begin
wake up
ProcessFunc111 end
workerThreadHandler end task
tasks begin size:1 stop:0
140563906656000 tasks end size:0
ProcessFunc666 begin
ProcessFunc222 end
wake up
workerThreadHandler end task
ProcessFunc333ProcessFunc555ProcessFunc444 endendworkerThreadHandler end taskworkerThreadHandler end taskend
workerThreadHandler end task
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
wake up
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
ProcessFunc666 end
workerThreadHandler end task
tasks begin size:0 stop:0
wake up
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
...

测试程序2,调用线程池停止程序

#include <iostream>
#include <chrono>
#include <mutex>
#include "ThreadPool.h"using namespace std;static std::mutex m_mutex;
void ProcessFunc111()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc222()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc333()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc444()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc555()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc666()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}int main()
{ThreadPool::getInstance(5);ThreadPool::getInstance()->addFunction([](){ProcessFunc111();});ThreadPool::getInstance()->addFunction([](){ProcessFunc222();});ThreadPool::getInstance()->addFunction([](){ProcessFunc333();});ThreadPool::getInstance()->addFunction([](){ProcessFunc444();});ThreadPool::getInstance()->addFunction([](){ProcessFunc555();});ThreadPool::getInstance()->addFunction([](){ProcessFunc666();});std::this_thread::sleep_for(std::chrono::seconds(1));std::cout<<"stop all "<<std::endl;ThreadPool::getInstance()->stopAll(false);getchar();return 0;
}

执行结果:

tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:ThreadPool0 stop:0tasks begin size:0 stop:0
140190941017856 tasks end size:5
ProcessFunc111 begin
140190932625152 tasks end size:4
ProcessFunc222 begin
140190966195968 tasks end size:3
ProcessFunc333 begin
140190949410560 tasks end size:2
ProcessFunc444 begin
140190957803264 tasks end size:1
ProcessFunc555 begin
stop all
wake up
ProcessFunc111 end
workerThreadHandler end task
ProcessFunc444 end
workerThreadHandler end task
ProcessFunc333 end
workerThreadHandler end task
ProcessFunc222 end
workerThreadHandler end task
ProcessFunc555 end
workerThreadHandler end task

测试程序3,立即停止线程池

#include <iostream>
#include <chrono>
#include <mutex>
#include "ThreadPool.h"using namespace std;static std::mutex m_mutex;
void ProcessFunc111()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc222()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc333()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc444()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc555()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc666()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}int main()
{ThreadPool::getInstance(5);ThreadPool::getInstance()->addFunction([](){ProcessFunc111();});ThreadPool::getInstance()->addFunction([](){ProcessFunc222();});ThreadPool::getInstance()->addFunction([](){ProcessFunc333();});ThreadPool::getInstance()->addFunction([](){ProcessFunc444();});ThreadPool::getInstance()->addFunction([](){ProcessFunc555();});ThreadPool::getInstance()->addFunction([](){ProcessFunc666();});std::this_thread::sleep_for(std::chrono::seconds(1));std::cout<<"stop all "<<std::endl;ThreadPool::getInstance()->stopAll(true);getchar();return 0;
}

执行结果:

tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
ThreadPool
139831215929088 tasks end size:5
ProcessFunc111 begin
tasks begin size:5 stop:0
139831199143680 tasks end size:4
ProcessFunc222 begin
139831224321792 tasks end size:3
ProcessFunc333 begin
139831207536384 tasks end size:2
ProcessFunc444 begin
139831232714496 tasks end size:1
ProcessFunc555 begin
stop all

相关文章:

C++简易线程池

原理说明&#xff1a; 1. 线程池创建时&#xff0c;指定线程池的大小thread_size。当有新的函数任务通过函数addFunction ()添加进来后&#xff0c;其中一个线程执行函数。一个线程一次执行一个函数。如果函数数量大与线程池数量&#xff0c;则后来的函数等待。 2. 线程池内部…...

【MATLAB】PSO粒子群优化LSTM(PSO_LSTM)的时间序列预测

有意向获取代码&#xff0c;请转文末观看代码获取方式~也可转原文链接获取~ 1 基本定义 PSO粒子群优化LSTM&#xff08;PSO-LSTM&#xff09;是一种将粒子群优化算法&#xff08;PSO&#xff09;与长短期记忆神经网络&#xff08;LSTM&#xff09;相结合的混合模型。该算法通过…...

产品经理学习-怎么写PRD文档

目录 瀑布流方法论介绍 产品需求文档&#xff08;PRD&#xff09;介绍 产品需求文档的基本要素 撰写产品需求文档 优先产品需求文档的特点 其他相关文档 瀑布流方法论介绍 瀑布流模型是一种项目的开发和管理的方法论&#xff0c;是敏捷的开发管理方式相对应的另一种方法…...

第3课 获取并播放音频流

本课对应源文件下载链接&#xff1a; https://download.csdn.net/download/XiBuQiuChong/88680079 FFmpeg作为一套庞大的音视频处理开源工具&#xff0c;其源码有太多值得研究的地方。但对于大多数初学者而言&#xff0c;如何快速利用相关的API写出自己想要的东西才是迫切需要…...

Spark编程实验四:Spark Streaming编程

目录 一、目的与要求 二、实验内容 三、实验步骤 1、利用Spark Streaming对三种类型的基本数据源的数据进行处理 2、利用Spark Streaming对Kafka高级数据源的数据进行处理 3、完成DStream的两种有状态转换操作 4、把DStream的数据输出保存到文本文件或MySQL数据库中 四…...

Flink去重计数统计用户数

1.数据 订单表&#xff0c;分别是店铺id、用户id和支付金额 "店铺id,用户id,支付金额", "shop-1,user-1,1", "shop-1,user-2,1", "shop-1,user-2,1", "shop-1,user-3,1", "shop-1,user-3,1", "shop-1,user…...

力扣:62. 不同路径(动态规划,附python二维数组的定义)

题目&#xff1a; 一个机器人位于一个 m x n 网格的左上角 &#xff08;起始点在下图中标记为 “Start” &#xff09;。 机器人每次只能向下或者向右移动一步。机器人试图达到网格的右下角&#xff08;在下图中标记为 “Finish” &#xff09;。 问总共有多少条不同的路径&…...

2022年全球运维大会(GOPS深圳站)-核心PPT资料下载

一、峰会简介 GOPS 主要面向运维行业的中高端技术人员&#xff0c;包括运维、开发、测试、架构师等群体。目的在于帮助IT技术从业者系统学习了解相关知识体系&#xff0c;让创新技术推动社会进步。您将会看到国内外知名企业的相关技术案例&#xff0c;也能与国内顶尖的技术专家…...

8868体育助力意甲罗马俱乐部 迪巴拉有望付出

8868体育助力意甲罗马俱乐部 迪巴拉有望付出 意甲罗马俱乐部是8868体育合作球队之一&#xff0c;本赛季&#xff0c;在意甲第14轮的比赛中&#xff0c;罗马客场2-1战胜萨索洛&#xff0c;积分上升到意甲第4位。 有报道称&#xff0c;迪巴拉在对阵佛罗伦萨的比赛中受伤&#xff…...

java设计模式实战【策略模式+观察者模式+命令模式+组合模式,混合模式在支付系统中的应用】

引言 在代码开发的世界里&#xff0c;理论知识的重要性毋庸置疑&#xff0c;但实战经验往往才是知识的真正试金石。正所谓&#xff0c;“读万卷书不如行万里路”&#xff0c;理论的学习需要通过实践来验证和深化。设计模式作为软件开发中的重要理论&#xff0c;其真正的价值在…...

小程序wx:if 和hidden的区别?

在小程序中&#xff0c;wx:if 和 hidden 是用于条件渲染的两种不同方式。 选择使用哪种方式取决于具体情况。如果条件变化频繁或节点包含复杂的子节点&#xff0c;可以考虑使用 wx:if 进行条件渲染&#xff1b;如果条件变化较少且节点结构简单&#xff0c;可以使用 hidden 控制…...

自动驾驶学习笔记(二十三)——车辆控制模型

#Apollo开发者# 学习课程的传送门如下&#xff0c;当您也准备学习自动驾驶时&#xff0c;可以和我一同前往&#xff1a; 《自动驾驶新人之旅》免费课程—> 传送门 《Apollo开放平台9.0专项技术公开课》免费报名—>传送门 文章目录 前言 运动学模型 动力学模型 总结…...

Linux Shell 015-文本双向覆盖重定向工具tee

Linux Shell 015-文本双向覆盖重定向工具tee 本节关键字&#xff1a;Linux、Bash Shell、文本双向覆盖重定向工具 相关指令&#xff1a;tee、echo、cat tee介绍 tee工具是从标准输入读取并写入到标准输出和文件&#xff0c;即&#xff1a;双向覆盖重定向&#xff08;屏幕输出…...

【PyQt】(自定义类)QIcon派生,更易用的纯色Icon

嫌Qt自带的icon太丑&#xff0c;自己写了一个&#xff0c;主要用于纯色图标的自由改色。 当然&#xff0c;图标素材得网上找。 Qt原生图标与现代图标对比&#xff1a; 没有对比就没有伤害 Qt图标 网络素材图标 自定义类XJQ_Icon&#xff1a; from PyQt5.QtGui import QIc…...

【mysql】数据处理格式化、转换、判断

数据处理 判断是否超时&#xff0c;时间是否大于当前时间计算分钟数时间格式化处理如果数值类型进行转换字符类型字符拼接case-when代替if-else判断数据空&#xff08;特殊&#xff1a;含空数据、空字符处理&#xff09; select /*判断是否超时&#xff0c;时间是否大于当前…...

深入探索Java中的UDP网络通信机制

在网络通信中&#xff0c;UDP&#xff08;User Datagram Protocol&#xff0c;用户数据报协议&#xff09;是一种无连接的协议&#xff0c;它在某些情况下比TCP更适合&#xff0c;尤其是在要求速度快、对数据准确性要求相对较低的场景下。本文将介绍如何使用Java进行UDP网络通信…...

List常见方法和遍历操作

List集合的特点 有序&#xff1a; 存和取的元素顺序一致有索引&#xff1a;可以通过索引操作元素可重复&#xff1a;存储的元素可以重复 List集合的特有方法 Collection的方法List都继承了List集合因为有索引&#xff0c;所以有了很多操作索引的方法 ublic static void main…...

【基础篇】一、认识JVM

文章目录 1、虚拟机2、Java虚拟机3、JVM的整体结构4、Java代码的执行流程5、JVM的三大功能6、JVM的分类7、JVM的生命周期 1、虚拟机 虚拟机&#xff0c;Virtual Machine&#xff0c;一台虚拟的计算机&#xff0c;用来执行虚拟计算机指令。分为&#xff1a; 系统虚拟机&#x…...

DrGraph原理示教 - OpenCV 4 功能 - 颜色空间

前言 前段时间&#xff0c;甲方提出明确需求&#xff0c;让把软件国产化。稍微研究了一下&#xff0c;那就转QT开发&#xff0c;顺便把以前的功能代码重写一遍。 至于在Ubuntu下折腾QT、OpenCV安装事宜&#xff0c;网上文章很多&#xff0c;照猫画虎即可。 这个过程&#xff0…...

听GPT 讲Rust源代码--src/tools(36)

File: rust/src/tools/clippy/clippy_lints/src/loops/empty_loop.rs 在Rust源代码中&#xff0c;empty_loop.rs文件位于src/tools/clippy/clippy_lints/src/loops/目录下&#xff0c;它的作用是实现并提供一个名为EMPTY_LOOP的Lint规则。Clippy是一个Rust的静态分析工具&#…...

一天一个开源项目(第56篇):人人都能用英语 - AI 时代的外语学习开源项目

引言 “其实一个字就够了&#xff1a;用。” 这是「一天一个开源项目」系列的第 56 篇文章。今天介绍的项目是 人人都能用英语&#xff08;GitHub&#xff09;。 学英语的核心是什么&#xff1f;李笑来在 2010 年的著作里用一个字概括&#xff1a;用。如今&#xff0c;这个经典…...

Turbo实战:如何用任务编排优化你的Monorepo构建流程?以pnpm+vitepress为例

Turbo实战&#xff1a;如何用任务编排优化你的Monorepo构建流程&#xff1f;以pnpmvitepress为例 在当今前端工程化领域&#xff0c;Monorepo已成为管理复杂项目的标配方案。但当项目规模增长到一定程度时&#xff0c;传统的构建方式往往会面临效率瓶颈——每次全量构建耗时漫长…...

Aspose.Words避坑指南:Java实现Word转PDF时如何去除水印(2023最新版)

Aspose.Words商业应用实战&#xff1a;Java版Word转PDF无水印解决方案深度解析 在企业级文档处理系统中&#xff0c;Word到PDF的转换需求几乎无处不在——合同归档、报告生成、电子发票导出等场景都依赖这一基础功能。作为Java开发者&#xff0c;当我们选择Aspose.Words这一业界…...

WikiJS全文搜索实战:用ElasticSearch+IK分词器提升内容检索效率(Docker版)

WikiJS全文搜索实战&#xff1a;ElasticSearch与IK分词器的深度优化指南 引言&#xff1a;为什么需要专业级全文搜索解决方案&#xff1f; 想象一下&#xff0c;当你面对一个包含数千篇技术文档的Wiki系统时&#xff0c;传统的关键词匹配就像在黑暗房间里寻找一根针。WikiJS自带…...

CLIP-GmP-ViT-L-14与YOLOv11结合:实现目标检测后的细粒度语义描述

CLIP-GmP-ViT-L-14与YOLOv11结合&#xff1a;实现目标检测后的细粒度语义描述 你有没有遇到过这种情况&#xff1f;一个智能摄像头告诉你“画面里有人”&#xff0c;但你更想知道的是“画面里有一个穿着蓝色外套、正在打电话的年轻人”。或者&#xff0c;一个货架分析系统告诉…...

从一道经典OJ题出发:详解二叉树‘凹入表示法’的输出技巧与C++实现

从一道经典OJ题出发&#xff1a;详解二叉树‘凹入表示法’的输出技巧与C实现 1. 凹入表示法的独特魅力与实现挑战 在算法竞赛和数据结构面试中&#xff0c;二叉树的输出格式往往成为区分选手水平的关键细节。不同于常见的层序遍历或图形化展示&#xff0c;凹入表示法&#xff0…...

【STM32实战】步进电机S型曲线算法优化与误差补偿策略

1. 为什么需要S型曲线算法 我第一次用步进电机做项目时&#xff0c;直接给电机发固定频率的脉冲让它转起来。结果电机启动瞬间发出"咔咔"的异响&#xff0c;运行起来也一顿一顿的。后来才知道&#xff0c;步进电机最怕的就是突然加速或急停&#xff0c;这会导致丢步、…...

AI 模型推理框架性能分析与对比

AI模型推理框架性能分析与对比 随着人工智能技术的快速发展&#xff0c;AI模型推理框架成为支撑各类应用落地的核心工具。无论是计算机视觉、自然语言处理还是推荐系统&#xff0c;高效的推理框架直接影响模型的响应速度、资源占用和部署成本。本文将从多个维度对比主流AI推理…...

老旧Mac如何重获新生?OCLP-Mod带来的系统升级解决方案

老旧Mac如何重获新生&#xff1f;OCLP-Mod带来的系统升级解决方案 【免费下载链接】OCLP-Mod A mod version for OCLP,with more interesting features. 项目地址: https://gitcode.com/gh_mirrors/oc/OCLP-Mod 随着科技的快速迭代&#xff0c;许多曾经性能卓越的Mac设备…...

Flink技术实践-超时异常踩坑与优化

一、背景介绍在Flink实时计算的生产环境中&#xff0c;最令人头疼的往往不是复杂的业务逻辑&#xff0c;而是那些突如其来的“超时异常”。这些异常就像是系统中的“幽灵”&#xff0c;通常在业务高峰期或网络抖动时出现&#xff0c;导致作业重启、数据延迟甚至数据丢失。最近几…...