python线程池管理两个队列_Python 线程池使用有限大小的工作队列

在去年的一篇 Python 多线程编程中学习了 Python 中如何使用多线程来调度任务，工作中也不时从自己的博客中找来参考。在运用当中不时的碰到内存消耗殆尽情况，直接把命令行窗口打死，不得不强行关窗口或杀进程。之前一直未意识到问题所在，只知任务太多就必死无疑，现在要用 Python 来处理大量任务了，必须着手来解决一下它。其实原因很简单，和 Java 的 ThreadPoolExecutor 一样(看它们用的类名都是一样的)。Java 的 ThreadPoolExecutor 内部使用了一个 Integer.MAX_VALUE 的 LinkedBlockingQueue 来存放提交的待处理的任务，所以基本上就是一个无底洞，自然解决办法也是类似的，需要一个 Bounded Queue 来存放任务列表。

在解决该问题之前自己也不妨来温习一下 Python 中使用线程池的基本模式，下面的模板代码曾经是我的最爱：

importtime

fromconcurrent.futuresimportThreadPoolExecutor

defperform(x):

time.sleep(2)

print(f'process {x}')

returnx+1

withThreadPoolExecutor(5)asexecutor:

foriinrange(3):

executor.submit(perform,i)

print('done')

在上面的 with 上下文中会进行以下几步

创建线程池

提交任务到线程池

等待所有任务完成

最后关闭线程池，并执行后面的代码

所以执行后的输出大概如下：

process 2

process 0

process 1

done

前三行的输出顺序不定，它们由线程池中的线程执行的，done 一定是在所有任务完成了最后输出的。

如果要收集子任务的输出，可以提交任务时放到 futures 列表中，如

futures.append(executor.submit(perform,i))

forfuturesinfutures:

print(future.result)

或是 executor.map() 一步提交多个任务并收集结果

results=executor.submit(perform,[0,1,2])

forresultinresults:

print(result)

回顾完了我们重新回到正题上来，如果不停的向线程池提交任务，待执行的任务全部要积压在线程池的工作队列中，提交多了快了，远远超出了线程池的处理速度就会迅速把本地内存挤暴。来观察一下一个 ThreadPoolExecutor 的内部属性

注意到它有一个 _work_queue 用来存放通过 submit 提交的待处理任务，默认实现为 SimpleQueue, 而 queue.SimpleQueue 是一个没有节制的队列，你可以一直往里面添加记录，只要计数没溢出并且内存充足。这就是简单使用 ThreadPoolExecutor 造成 OutOfMemory 的元凶，更可恶的是 Python 不坦诚的告诉我们内存不足，而是直接把系统拖死。

因此解决办法就是要想法设法在提交任务之前检查当前 _work_queue 大小是否超过某个限定值，是的话等降下来后再提交新任务。然而 _work_queue 是一个私有变量，不允许从外部访问，那么索性我们创造一个子类把 _work_queue 置换成一个有容量限制的 Queue, 那就是 queue.Queue. BoundedThreadPoolExecutor 并使用方式如下：

importtime

fromconcurrent.futuresimportThreadPoolExecutor

fromqueueimportQueue

defperform(x):

time.sleep(2)

print(f'process {x}')

classBoundedThreadPoolExecutor(ThreadPoolExecutor):

def__init__(self,max_workers,max_waiting_tasks,*args,**kwargs):

super().__init__(max_workers=max_workers,*args,**kwargs)

self._work_queue=Queue(maxsize=max_waiting_tasks)

withBoundedThreadPoolExecutor(10,100)asexecutor:

foriinrange(999999999):

executor.submit(perform,i)

print('done')

现在操作系统再也不用担心过多的任务会把内存消耗干净了。再来看一下 BoundedThreadPoolExecutor 内部

_work_queue 变成了 queue.Queue 实现，原理是借助于 Queue 可指定 maxsize，当 Queue 的大小达到这个值，提交任务时将被阻塞，直致工作线程从中取走待执行任务，Queue 的大小低于 maxsize 值，才能继续往里头提交任务，这达到了一种流量控制的效果。

链接：

python线程池管理两个队列_Python 线程池使用有限大小的工作队列相关推荐

python线程退出或应用程序请求_Python 线程和进程
前言学编程,谁没有为线程折腾过啊. 目录线程与进程线程与进程是操作系统里面的术语,简单来讲,每一个应用程序都有一个自己的进程. 操作系统会为这些进程分配一些执行资源,例如内存空间等. 在进程中, ...
python如何在所有线程结束后执行最后操作_Python线程的生命周期
在线程的生命周期中,从创建到执行以及最终终止,线程通常处于四种状态之一:开始态.可调度状态.阻塞态和终止态. 父线程和子线程当一个新的进程或程序开始运行时,它将以一个线程开始,这个线程被称为主线程. ...
python用链表求两数之和_python 算法 - 008 计算两个链表所代表的整数之和 (整数相加法)...
python 算法 - 008 计算两个链表所代表的整数之和 (整数相加法) 知之者不如好之者, 好之者不如乐之者.--<雍也> 知道德者不如好道德者, 好道德者不如乐道德者, 是为形容人 ...
Linux c 线程分离（两种方法实现线程分离）
文章目录 1.使用函数实现线程分离 2.通过属性实现线程分离 3.杀死线程函数 1.使用函数实现线程分离 // 创建线程 int pthread_create(pthread_t &tid,c ...
python线程池传入多个参数_python 线程池ThreadPoolExecutor
背景最近项目中需要抽取数据库中某些数据组成一个缓存表,SQL倒是不复杂就是单纯的慢,慢到navicat跑崩了也没跑出来,然后就想着用python写脚本,根据时间字段做异步查询代码数据库连接写成伪 ...
python线程池传入多个参数_python线程池问题
创建10个线程, 然后依次读取线程对象的值 from threading import current_thread as cthread from concurrent.futures import ...
python 线程通信的几种方式_Python 线程、线程通信、多线程
这是一篇学习Python 线程相关的内容,记录一下以备复习和开发使用,技术有限,如有问题欢迎指出,多谢. 一.GIL 全局解释器锁(cpython) 1.为什么会有这个锁:为了线程安全,减少pytho ...
python 函数参数前面两个星号_Python中参数前面一个星号两个星号(*参数，**参数)起什么作用呢？...
摘要: 下文讲述Python中参数前面一个星号两个星号的功能分享,如下所示: 在Python语言中,我们经常看见参数前面加上一个星号或两个星号那么这些写法到底起什么作用呢? 下文将一一道来,如下所 ...
python format函数保留两位小数_python format函数
在Python 3.0中,%操作符通过一个更强的格式化方法format()进行了增强.对str.format()的支持已经被反向移植到了Python 2.6在2.6中,8-bit字符串和Unicode ...
python怎样连续输入两个数字_python实现输入数字的连续加减
不用库,写了很久,一直出bug,到网上一搜,可以直接输入之后,eval(str)即可得到结果! eval程序如下: s=input("请输入要运算的数字") print(" ...

python线程池管理两个队列_Python 线程池使用有限大小的工作队列

python线程池管理两个队列_Python 线程池使用有限大小的工作队列相关推荐

最新文章

热门文章