python multiprocessing进程超时处理与进度条显示的平衡

本文主要是介绍python multiprocessing进程超时处理与进度条显示的平衡,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

最近写python项目,遇到几个多进程的问题

多进程情况下,如何显示总任务的运行进度

用的比较多的是tqdm库,使用非常方便,在有迭代的地方加上tqdm即可。

比如使用Pool.imap()执行任务时

from multiprocessing import Pool
with Pool(4) as p:res = list(tqdm(p.imap(function_name, params), total=len(params)))
p.close()
p.join()

或者通过for循环执行任务时

from multiprocessing import Pool
pool = Pool(processes=4)
for i  in tqdm(range(500)):pool.apply(test, args=(i,))   
pool.close()
pool.join()

我的项目中需要对大量的url进行请求和处理,所以经常会出现某个进程的请求长时间没有响应而被挂起的情况,导致tqdm经常会卡在99%或100%处无法结束。tqdm本身似乎也没有超时机制,所以在利用多进程处理爬虫这类需要进程等待结果的任务时,必须要添加子进程的超时机制。

多进程的超时处理

我研究了一下进程池超时处理的几个办法。比较主流的有以下几种:

  1. get获取多进程的返回值,获取超时则抛出异常
  2. 在需要执行的任务函数中添加计时器和时间,超时抛出异常
  3. 为每个任务进程添加守护进程,由守护进程计算执行的时间,如果超时则守护进程杀死任务进程

多进程超时与总进度条的显示

将多进程超时处理与进度条处理结合了一下

  • 方法一
    使用map_async处理任务,通过get超时判断进程超时。这个方法实现比较简单,但是运行起来非常的慢。
pool = Pool(processes=4)
for l in tqdm(links,total=1000): # total为links的总数r = pool.map_async(self.check,(l,))try:print(r.get(timeout=3)) # 设置子进程的响应超时为3sexcept multiprocessing.context.TimeoutError:pass
  • 方法二
    使用imap_unordered()执行任务,进程执行的结果会在进程执行完后就立刻返回,不像map()需要等待所有任务执行完一次性返回结果。这里可以通过next迭代获取该函数所有进程的下的返回值,当获取超时,则抛出超时异常。但是测试发现,如果进程返回超时,.next会重复获取超时进程的返回结果,重复抛出超时异常,而不是在抛出超时异常后执行下一个任务,也相当于是挂起状态了,并没有解决进程超时的问题。看了一下imap_unordered()的源代码,发现next的逻辑是在没有获取到数据之前,会一直重复请求直到取到数据为止。不知道是不是还有别的隐藏技能,总觉得这个逻辑设计的怪怪的。暂时也不知道怎么让.next只获取一次,获取失败就跳到下一个index,所以这个方案只能作罢。
with multiprocessing.Pool() as pool:it = pool.imap_unordered(self.check, links) #check为任务函数pbar = tqdm(total=len(links)) #提前设置任务总量while True:try:tmp = it.next(timeout=2)pbar.update()except multiprocessing.TimeoutError:print("link timeout: {}".format(tmp))continueexcept StopIteration:# signal that the iterator is exhausted# 如果所有的进程数据都获取完了,next会抛出该异常停止获取pbar.close()break
  • 方法二改进版
    方案二中,.next一直会卡在超时的index上不继续迭代,所以不会有StopIteration异常使while循环停止,陷入死循环。所以想到一个比较简单粗暴的办法,将.next获取返回值的次数(即循环次数)限制为数组长度,避免陷入死循环。
    按照该办法可以正常进行超时处理了,但是测试发现这个办法存在一个bug:如果循环中有一个进程没有响应,则该进程后的所有进程返回值都无法获取到,全部被该进程的超时取代。因为.next函数会坚持不懈地去请求获取这个超时进程的数据,导致其他进程的数据没办法获取。emmm…这个办法也只能作罢。

  • 方法三
    这个是看stackoverflow的大佬们提出的一个方案。建立一个守护进程池。由守护进程新建一个执行任务的进程,并监控执行进程的返回值,如果返回超时,则杀掉执行进程。这个是一圈下来的最佳方案,但是该方案与tqdm冲突,如果使用了tqdm,那么进程函数就不会执行(还没找到原因,哪位大佬知道的可以指点一下),所以需要自己另外写进度显示

# 执行函数
def check(self, url):headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:73.0) Gecko/20100101 Firefox/73.0","Connection": "close"}try:res = requests.head(url, headers=headers)return url, res.status_codeexcept Exception as e:print(e)# 守护进程
def abortable_check(self, func, *args, **kwargs):timeout = kwargs.get('timeout', None)p = ThreadPool(1)res = p.apply_async(func, args=args)try:out = res.get(timeout)  # Wait timeout seconds for func to complete.# print(out)return outexcept multiprocessing.TimeoutError:print("{} timeout".format(args[0]))return (args[0], 444)# 回调函数
def collectMyResult(self, result):if result is not None and result[1] == 404:self.brokenlinks.append(result[0])if result is not None and result[1] == 444:self.timeout.append(result[0])self.count += 1 #全局变量统计任务执行进度print('\r进度:{}/{}'.format(self.count,self.linkslen), end ='')def main():pool = multiprocessing.Pool(maxtasksperchild=10)for l in links:a = [l]abortable_func = partial(self.abortable_check, self.check, timeout=5)pool.apply_async(abortable_func, args=a, callback=self.collectMyResult)

这篇关于python multiprocessing进程超时处理与进度条显示的平衡的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/752891

相关文章

Python通用唯一标识符模块uuid使用案例详解

《Python通用唯一标识符模块uuid使用案例详解》Pythonuuid模块用于生成128位全局唯一标识符,支持UUID1-5版本,适用于分布式系统、数据库主键等场景,需注意隐私、碰撞概率及存储优... 目录简介核心功能1. UUID版本2. UUID属性3. 命名空间使用场景1. 生成唯一标识符2. 数

Python办公自动化实战之打造智能邮件发送工具

《Python办公自动化实战之打造智能邮件发送工具》在数字化办公场景中,邮件自动化是提升工作效率的关键技能,本文将演示如何使用Python的smtplib和email库构建一个支持图文混排,多附件,多... 目录前言一、基础配置:搭建邮件发送框架1.1 邮箱服务准备1.2 核心库导入1.3 基础发送函数二、

Javaee多线程之进程和线程之间的区别和联系(最新整理)

《Javaee多线程之进程和线程之间的区别和联系(最新整理)》进程是资源分配单位,线程是调度执行单位,共享资源更高效,创建线程五种方式:继承Thread、Runnable接口、匿名类、lambda,r... 目录进程和线程进程线程进程和线程的区别创建线程的五种写法继承Thread,重写run实现Runnab

Python包管理工具pip的升级指南

《Python包管理工具pip的升级指南》本文全面探讨Python包管理工具pip的升级策略,从基础升级方法到高级技巧,涵盖不同操作系统环境下的最佳实践,我们将深入分析pip的工作原理,介绍多种升级方... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

基于Python实现一个图片拆分工具

《基于Python实现一个图片拆分工具》这篇文章主要为大家详细介绍了如何基于Python实现一个图片拆分工具,可以根据需要的行数和列数进行拆分,感兴趣的小伙伴可以跟随小编一起学习一下... 简单介绍先自己选择输入的图片,默认是输出到项目文件夹中,可以自己选择其他的文件夹,选择需要拆分的行数和列数,可以通过

Python中反转字符串的常见方法小结

《Python中反转字符串的常见方法小结》在Python中,字符串对象没有内置的反转方法,然而,在实际开发中,我们经常会遇到需要反转字符串的场景,比如处理回文字符串、文本加密等,因此,掌握如何在Pyt... 目录python中反转字符串的方法技术背景实现步骤1. 使用切片2. 使用 reversed() 函

Python中将嵌套列表扁平化的多种实现方法

《Python中将嵌套列表扁平化的多种实现方法》在Python编程中,我们常常会遇到需要将嵌套列表(即列表中包含列表)转换为一个一维的扁平列表的需求,本文将给大家介绍了多种实现这一目标的方法,需要的朋... 目录python中将嵌套列表扁平化的方法技术背景实现步骤1. 使用嵌套列表推导式2. 使用itert

使用Docker构建Python Flask程序的详细教程

《使用Docker构建PythonFlask程序的详细教程》在当今的软件开发领域,容器化技术正变得越来越流行,而Docker无疑是其中的佼佼者,本文我们就来聊聊如何使用Docker构建一个简单的Py... 目录引言一、准备工作二、创建 Flask 应用程序三、创建 dockerfile四、构建 Docker

Python使用vllm处理多模态数据的预处理技巧

《Python使用vllm处理多模态数据的预处理技巧》本文深入探讨了在Python环境下使用vLLM处理多模态数据的预处理技巧,我们将从基础概念出发,详细讲解文本、图像、音频等多模态数据的预处理方法,... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

Python使用pip工具实现包自动更新的多种方法

《Python使用pip工具实现包自动更新的多种方法》本文深入探讨了使用Python的pip工具实现包自动更新的各种方法和技术,我们将从基础概念开始,逐步介绍手动更新方法、自动化脚本编写、结合CI/C... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核