python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池

目录
  • 1、ThreadPoolExecutor多线程
    • <1>为什么需要线程池呢?
    • <2>标准库concurrent.futures模块
    • <3>简单使用
  • <4>as_completed(一次性获取所有的结果)
    • <5>map()方法
    • <6>wait()方法
  • 2、ProcessPoolExecutor多进程
    • <1>同步调用方式: 调用,然后等返回值,能解耦,但是速度慢
    • <2>异步调用方式:只调用,不等返回值,可能存在耦合,但是速度快
    • <3>怎么使用异步调用方式,但同时避免耦合的问题?
  • 3、总结

1、ThreadPoolExecutor多线程

<1>为什么需要线程池呢?

  • 对于io密集型,提高执行的效率。
  • 线程的创建是需要消耗系统资源的。

所以线程池的思想就是:每个线程各自分配一个任务,剩下的任务排队等待,当某个线程完成了任务的时候,排队任务就可以安排给这个线程继续执行。

<2>标准库concurrent.futures模块

它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类,
分别实现了对threading模块和multiprocessing模块的进一步抽象。

不仅可以帮我们自动调度线程,还可以做到:

  • 主线程可以获取某一个线程(或者任务)的状态,以及返回值
  • 当一个线程完成的时候,主线程能够立即知道
  • 让多线程和多进程的编码接口一致

<3>简单使用

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
import time

# 参数times用来模拟网络请求时间
def get_html(times):
    print("get page {}s finished".format(times))
   return times
# 创建线程池
# 设置线程池中最多能同时运行的线程数目,其他等待
executor = ThreadPoolExecutor(max_workers=2)
# 通过submit函数提交执行的函数到线程池中,submit函数立即返回,不阻塞
# task1和task2是任务句柄
task1 = executor.submit( get_html, (2) )
task2 = executor.submit( get_html, (3) )

# done()方法用于判断某个任务是否完成,bool型,完成返回True,没有完成返回False
print( task1.done() )
# cancel()方法用于取消某个任务,该任务没有放到线程池中才能被取消,如果已经放进线程池子中,则不能被取消
# bool型,成功取消了返回True,没有取消返回False
print( task2.cancel() )
# result()方法可以获取task的执行结果,前提是get_html()函数有返回值
print( task1.result() )
print( task2.result() )
# 结果:
# get page 3s finished
# get page 2s finished
# True
# False

# 2
# 3

ThreadPoolExecutor类在构造实例的时候,传入max_workers参数来设置线程池中最多能同时运行的线程数目
使用submit()函数来提交线程需要执行任务(函数名和参数)到线程池中,并返回该任务的句柄,

注意:submit()不是阻塞的,而是立即返回。

通过submit()函数返回的任务句柄,能够使用done()方法判断该任务是否结束,使用cancel()方法来取消,使用result()方法可以获取任务的返回值,查看内部代码,发现该方法是阻塞的

<4>as_completed(一次性获取所有的结果)

上面虽然提供了判断任务是否结束的方法,但是不能在主线程中一直判断,有时候我们是得知某个任务结束了,就去获取结果,而不是一直判断每个任务有没有结束。这时候就可以使用as_completed方法一次取出所有任务的结果。

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

# 参数times用来模拟网络请求时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times

# 创建线程池子
# 设置最多2个线程运行,其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
# 一次性把所有的任务都放进线程池,得到一个句柄,但是最多只能同时执行2个任务
all_task = [ executor.submit(get_html,(each_url)) for each_url in urls ] 

for future in as_completed( all_task ):
    data = future.result()
    print("in main:get page {}s success".format(data))

# 结果
# get page 2s finished
# in main:get page 2s success
# get page 3s finished
# in main:get page 3s success
# get page 4s finished
# in main:get page 4s success
# 从结果可以看到,并不是先传入哪个url,就先执行哪个url,没有先后顺序

<5>map()方法

除了上面的as_completed()方法,还可以使用execumap方法。但是有一点不同,使用map方法,不需提前使用submit方法,
map方法与python标准库中的map含义相同,都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls列表中的每个元素都执行get_html()函数,并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同,输出顺序和urls列表的顺序相同,就算2s的任务先执行完成,也会先打印出3s的任务先完成,再打印2s的任务完成

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,as_completed
import time
# 参数times用来模拟网络请求时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
# 创建线程池子
# 设置最多2个线程运行,其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
for result in executor.map(get_html, urls):
    print("in main:get page {}s success".format(result))

结果:

get page 2s finished
 get page 3s finished
 in main:get page 3s success
 in main:get page 2s success
 get page 4s finished
 in main:get page 4s success

<6>wait()方法

wait方法可以让主线程阻塞,直到满足设定的要求。wait方法接收3个参数,等待的任务序列、超时时间以及等待条件。
等待条件return_when默认为ALL_COMPLETED,表明要等待所有的任务都借宿。可以看到运行结果中,确实是所有任务都完成了,主线程才打印出main,等待条件还可以设置为FIRST_COMPLETED,表示第一个任务完成就停止等待。

超时时间参数可以不设置:

wait()方法和as_completed(), map()没有关系。不管你是用as_completed(),还是用map()方法,你都可以在执行主线程之前使用wait()。
as_completed()和map()是二选一的。

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED,FIRST_COMPLETED
import time
# 参数times用来模拟网络请求时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times

# 创建线程池子
# 设置最多2个线程运行,其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
all_task = [executor.submit(get_html,(url)) for url in urls]
wait(all_task,return_when=ALL_COMPLETED)
print("main")
# 结果
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main

2、ProcessPoolExecutor多进程

<1>同步调用方式: 调用,然后等返回值,能解耦,但是速度慢

import datetime
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import time, random, os
import requests
def task(name):
    print('%s %s is running'%(name,os.getpid()))
    #print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

if __name__ == '__main__':
    p = ProcessPoolExecutor(4)  # 设置

    for i in range(10):
        # 同步调用方式,不仅要调用,还要等返回值
        obj = p.submit(task, "进程pid:")  # 传参方式(任务名,参数),参数使用位置或者关键字参数
        res = obj.result()
    p.shutdown(wait=True)  # 关闭进程池的入口,等待池内任务运行结束
    print("主")
################
################
# 另一个同步调用的demo
def get(url):
    print('%s GET %s' % (os.getpid(),url))
    time.sleep(3)
    response = requests.get(url)
    if response.status_code == 200:
        res = response.text
    else:
        res = "下载失败"
    return res  # 有返回值

def parse(res):
    time.sleep(1)
    print("%s 解析结果为%s" %(os.getpid(),len(res)))

if __name__ == "__main__":
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
    ]
    p=ProcessPoolExecutor(9)
    l=[]
    start = time.time()
    for url in urls:
        future = p.submit(get,url)  # 需要等结果,所以是同步调用
        l.append(future)

    # 关闭进程池,等所有的进程执行完毕
    p.shutdown(wait=True)
    for future in l:
        parse(future.result())
    print('完成时间:',time.time()-start)
    #完成时间: 13.209137678146362

<2>异步调用方式:只调用,不等返回值,可能存在耦合,但是速度快

def task(name):
    print("%s %s is running" %(name,os.getpid()))
    time.sleep(random.randint(1,3))
if __name__ == '__main__':
    p = ProcessPoolExecutor(4) # 设置进程池内进程
    for i in range(10):
        # 异步调用方式,只调用,不等返回值
        p.submit(task,'进程pid:') # 传参方式(任务名,参数),参数使用位置参数或者关键字参数
    p.shutdown(wait=True)  # 关闭进程池的入口,等待池内任务运行结束
    print('主')
##################
##################
# 另一个异步调用的demo
def get(url):
    print('%s GET %s' % (os.getpid(),url))
    time.sleep(3)
    reponse = requests.get(url)
    if reponse.status_code == 200:
        res = reponse.text
    else:
        res = "下载失败"
    parse(res)  # 没有返回值
def parse(res):
    time.sleep(1)
    print('%s 解析结果为%s' %(os.getpid(),len(res)))

if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',

    ]
    p = ProcessPoolExecutor(9)
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
    p.shutdown(wait=True)
    print("完成时间",time.time()-start)#  完成时间 6.293345212936401

<3>怎么使用异步调用方式,但同时避免耦合的问题?

(1)进程池:异步 + 回调函数,,cpu密集型,同时执行,每个进程有不同的解释器和内存空间,互不干扰

def get(url):
    print('%s GET %s' % (os.getpid(), url))
    time.sleep(3)
    response = requests.get(url)
    if response.status_code == 200:
        res = response.text
    else:
        res = '下载失败'
    return res
def parse(future):
    time.sleep(1)
    # 传入的是个对象,获取返回值 需要进行result操作
    res = future.result()
    print("res",)
    print('%s 解析结果为%s' % (os.getpid(), len(res)))
if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
    ]
    p = ProcessPoolExecutor(9)
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
        #模块内的回调函数方法,parse会使用future对象的返回值,对象返回值是执行任务的返回值
        #回调应该是相当于parse(future)
        future.add_done_callback(parse)
   p.shutdown(wait=True)
    print("完成时间",time.time()-start)#完成时间 33.79998469352722

(2)线程池:异步 + 回调函数,IO密集型主要使用方式,线程池:执行操作为谁有空谁执行

def get(url):
    print("%s GET %s" %(current_thread().name,url))
    time.sleep(3)
    reponse = requests.get(url)
    if reponse.status_code == 200:
        res = reponse.text
    else:
        res = "下载失败"
    return res
def parse(future):
    time.sleep(1)
    res = future.result()
    print("%s 解析结果为%s" %(current_thread().name,len(res)))
if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.sina.com.cn',
        'https://www.tmall.com',
        'https://www.jd.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://www.baidu.com',
        'https://www.baidu.com',
        'https://www.baidu.com',
    ]
    p = ThreadPoolExecutor(4)
    start = time.time()
    for url in urls:
        future = p.submit(get,url)
        future.add_done_callback(parse)
    p.shutdown(wait=True)
    print("主",current_thread().name)
    print("完成时间",time.time()-start)#完成时间 32.52604126930237

3、总结

  • 1、线程不是越多越好,会涉及cpu上下文的切换(会把上一次的记录保存)。
  • 2、进程比线程消耗资源,进程相当于一个工厂,工厂里有很多人,里面的人共同享受着福利资源,,一个进程里默认只有一个主线程,比如:开启程序是进程,里面执行的是线程,线程只是一个进程创建多个人同时去工作。
  • 3、线程里有GIL全局解锁器:不允许cpu调度
  • 4、计算密度型适用于多进程
  • 5、线程:线程是计算机中工作的最小单元
  • 6、进程:默认有主线程 (帮工作)可以多线程共存
  • 7、协程:一个线程,一个进程做多个任务,使用进程中一个线程去做多个任务,微线程
  • 8、GIL全局解释器锁:保证同一时刻只有一个线程被cpu调度

到此这篇关于python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池的文章就介绍到这了,更多相关python ThreadPoolExecutor 内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • Python线程池模块ThreadPoolExecutor用法分析

    本文实例讲述了Python线程池模块ThreadPoolExecutor用法.分享给大家供大家参考,具体如下: python3内置的有Threadingpool和ThreadPoolExecutor模块,两个都可以做线程池,当然ThreadPoolExecutor会更好用一些,而且也有ProcessPoolExecutor进程池模块,使用方法基本一致. 首先导入模块 from concurrent.futures import ThreadPoolExecutor 使用方法很简单,最常用的可能就

  • 解决python ThreadPoolExecutor 线程池中的异常捕获问题

    问题 最近写了涉及线程池及线程的 python 脚本,运行过程中发现一个有趣的现象,线程池中的工作线程出现问题,引发了异常,但是主线程没有捕获异常,还在发现 BUG 之前一度以为线程池代码正常返回. 先说重点 这里主要想介绍 python concurrent.futuresthread.ThreadPoolExecutor 线程池中的 worker 引发异常的时候,并不会直接向上抛起异常,而是需要主线程通过调用concurrent.futures.Future.exception(timeou

  • python3线程池ThreadPoolExecutor处理csv文件数据

    目录 背景 知识点 拓展 库 流程 实现代码 解释 背景 由于不同乙方对服务商业务接口字段理解不一致,导致线上上千万数据量数据存在问题,为了修复数据,通过 Python 脚本进行修改 知识点 Python3.线程池.pymysql.CSV 文件操作.requests 拓展 当我们程序在使用到线程.进程或协程的时候,以下三个知识点可以先做个基本认知 CPU 密集型.IO 密集型.GIL 全局解释器锁 库 pip3 install requests pip3 install pymysql 流程 实

  • python 多进程并行编程 ProcessPoolExecutor的实现

    使用 ProcessPoolExecutor from concurrent.futures import ProcessPoolExecutor, as_completed import random 斐波那契数列 当 n 大于 30 时抛出异常 def fib(n): if n > 30: raise Exception('can not > 30, now %s' % n) if n <= 2: return 1 return fib(n-1) + fib(n-2) 准备数组 nu

  • python线程池 ThreadPoolExecutor 的用法示例

    前言 从Python3.2开始,标准库为我们提供了 concurrent.futures 模块,它提供了 ThreadPoolExecutor (线程池)和ProcessPoolExecutor (进程池)两个类. 相比 threading 等模块,该模块通过 submit 返回的是一个 future 对象,它是一个未来可期的对象,通过它可以获悉线程的状态主线程(或进程)中可以获取某一个线程(进程)执行的状态或者某一个任务执行的状态及返回值: 主线程可以获取某一个线程(或者任务的)的状态,以及返

  • python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池

    目录 1.ThreadPoolExecutor多线程 <1>为什么需要线程池呢? <2>标准库concurrent.futures模块 <3>简单使用 <4>as_completed(一次性获取所有的结果) <5>map()方法 <6>wait()方法 2.ProcessPoolExecutor多进程 <1>同步调用方式: 调用,然后等返回值,能解耦,但是速度慢 <2>异步调用方式:只调用,不等返回值,可能存在

  • Python中的线程操作模块(oncurrent)

    目录 GIL锁 1. 创建线程的方式:直接使用Thread 2. 创建线程的方式:继承Thread 二.多线程与多进程 1. pid的比较 2. 开启效率的较量 3. 内存数据的共享问题 三.Thread类的其他方法 1. 代码示例 2. join方法 四.多线程实现socket 五.守护线程 1. 详细解释 2. 守护线程例 六.同步锁 1. 多个线程抢占资源的情况 2.同步锁的引用 3.实例 七.死锁与递归锁 1. 死锁 2. 递归锁(可重入锁)RLock 3.典型问题:科学家吃面 八.线程

  • 详解python中的线程与线程池

    线程 进程和线程 什么是进程? 进程就是正在运行的程序, 一个任务就是一个进程, 进程的主要工作是管理资源, 而不是实现功能 什么是线程? 线程的主要工作是去实现功能, 比如执行计算. 线程和进程的关系就像员工与老板的关系, 老板(进程) 提供资源 和 工作空间, 员工(线程) 负责去完成相应的任务 特点 一个进程至少由一个线程, 这一个必须存在的线程被称为主线程, 同时一个进程也可以有多个线程, 即多线程 当我们我们遇到一些需要重复执行的代码时, 就可以使用多线程分担一些任务, 进而加快运行速

  • 使用Python中的线程进行网络编程的入门教程

    引言 对于 Python 来说,并不缺少并发选项,其标准库中包括了对线程.进程和异步 I/O 的支持.在许多情况下,通过创建诸如异步.线程和子进程之类的高层模块,Python 简化了各种并发方法的使用.除了标准库之外,还有一些第三方的解决方案,例如 Twisted.Stackless 和进程模块.本文重点关注于使用 Python 的线程,并使用了一些实际的示例进行说明.虽然有许多很好的联机资源详细说明了线程 API,但本文尝试提供一些实际的示例,以说明一些常见的线程使用模式. 全局解释器锁 (G

  • 详解python中的线程

    Python中创建线程有两种方式:函数或者用类来创建线程对象. 函数式:调用 _thread 模块中的start_new_thread()函数来产生新线程. 类:创建threading.Thread的子类来包装一个线程对象. 1.线程的创建 1.1 通过thread类直接创建 import threading import time def foo(n): time.sleep(n) print("foo func:",n) def bar(n): time.sleep(n) prin

  • 解决Python中定时任务线程无法自动退出的问题

    python的线程有一个类叫Timer可以,用来创建定时任务,但是它的问题是只能运行一次,如果要重复执行,则只能在任务中再调用一次timer,但这样就存在新的问题了,就是在主进程退出后,不能正常退出子线程. from threading import Timer def scheduletaskwrap(): pritn "in task" Timer(10, scheduletaskwrap).start() Timer(10, scheduletaskwrap).start() 象

  • 详解C语言和Python中的线程混用

    问题 你有一个程序需要混合使用C.Python和线程, 有些线程是在C中创建的,超出了Python解释器的控制范围. 并且一些线程还使用了Python C API中的函数. 解决方案 如果你想将C.Python和线程混合在一起,你需要确保正确的初始化和管理Python的全局解释器锁(GIL). 要想这样做,可以将下列代码放到你的C代码中并确保它在任何线程被创建之前被调用. #include <Python.h> ... if (!PyEval_ThreadsInitialized()) { P

  • Python中使用subprocess库创建附加进程

    前言 subprocess库提供了一个API创建子进程并与之通信.这对于运行生产或消费文本的程序尤其有好处,因为这个API支持通过新进行的标准输入和输出通道来回传数据. 本篇,将详细介绍Python创建附加进行的库:subprocess. run(运行外部命令) subprocess库本身可以替换os.system(),os.spawnv()等函数.现在我们来通过subprocess库运行一个外部命令,但不采用os.system().示例如下: import subprocess complet

  • 深入解析Python中的线程同步方法

    同步访问共享资源 在使用线程的时候,一个很重要的问题是要避免多个线程对同一变量或其它资源的访问冲突.一旦你稍不留神,重叠访问.在多个线程中修改(共享资源)等这些操作会导致各种各样的问题:更严重的是,这些问题一般只会在比较极端(比如高并发.生产服务器.甚至在性能更好的硬件设备上)的情况下才会出现. 比如有这样一个情况:需要追踪对一事件处理的次数 counter = 0 def process_item(item): global counter ... do something with item

  • python中的线程threading.Thread()使用详解

    1. 线程的概念: 线程,有时被称为轻量级进程(Lightweight Process,LWP),是程序执行流的最小单元.一个标准的线程由线程ID,当前指令指针(PC),寄存器集合和堆栈组成.另外,线程是进程中的一个实体,是被系统独立调度和分派的基本单位,线程自己不拥有系统资源,只拥有一点儿在运行中必不可少的资源,但它可与同属一个进程的其它线程共享进程所拥有的全部资源. 2. threading.thread()的简单使用 2.1 python的thread模块是比较底层的模块,python的t

随机推荐