精华内容
下载资源
问答
  • 2022-04-13 18:03:45
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=20) as t:
        t.map(lambda args: download_ts(*args), [[ts_url, i, len(tss_url)] for i, ts_url in enumerate(tss_url)])
    
    更多相关内容
  • 主要介绍了解决python ThreadPoolExecutor 线程池中的异常捕获问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
  • Python ThreadPoolExecutor线程池

    千次阅读 2021-12-30 11:03:37
    Python中已经有了threading模块,为什么还需要线程池呢,线程池又是什么东西呢?以爬虫为例,需要控制同时爬取的线程数,例子中创建了20个线程,而同时只允许3个线程在运行,但是20个线程都需要创建和销毁,线程的...

    概念

    Python中已经有了threading模块,为什么还需要线程池呢,线程池又是什么东西呢?以爬虫为例,需要控制同时爬取的线程数,例子中创建了20个线程,而同时只允许3个线程在运行,但是20个线程都需要创建和销毁,线程的创建是需要消耗系统资源的,有没有更好的方案呢?其实只需要三个线程就行了,每个线程各分配一个任务,剩下的任务排队等待,当某个线程完成了任务的时候,排队任务就可以安排给这个线程继续执行。

    这就是线程池的思想(当然没这么简单),但是自己编写线程池很难写的比较完美,还需要考虑复杂情况下的线程同步,很容易发生死锁。从Python3.2开始,标准库为我们提供了concurrent.futures模块,它提供了ThreadPoolExecutorProcessPoolExecutor两个类,实现了对threadingmultiprocessing的进一步抽象(这里主要关注线程池),不仅可以帮我们自动调度线程,还可以做到:

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

    实例

    简单使用

    from concurrent.futures import ThreadPoolExecutor
    import time
    
    # 参数times用来模拟网络请求的时间
    def get_html(times):
        time.sleep(times)
        print("get page {}s finished".format(times))
        return times
    
    executor = ThreadPoolExecutor(max_workers=2)
    # 通过submit函数提交执行的函数到线程池中,submit函数立即返回,不阻塞
    task1 = executor.submit(get_html, (3))
    task2 = executor.submit(get_html, (2))
    # done方法用于判定某个任务是否完成
    print(task1.done())
    # cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
    print(task2.cancel())
    time.sleep(4)
    print(task1.done())
    # result方法可以获取task的执行结果
    print(task1.result())
    
    # 执行结果
    # False  # 表明task1未执行完成
    # False  # 表明task2取消失败,因为已经放入了线程池中
    # get page 2s finished
    # get page 3s finished
    # True  # 由于在get page 3s finished之后才打印,所以此时task1必然完成了
    # 3     # 得到task1的任务返回值
    • ThreadPoolExecutor构造实例的时候,传入max_workers参数来设置线程池中最多能同时运行的线程数目。
    • 使用submit函数来提交线程需要执行的任务(函数名和参数)到线程池中,并返回该任务的句柄(类似于文件、画图),注意submit()不是阻塞的,而是立即返回。
    • 通过submit函数返回的任务句柄,能够使用done()方法判断该任务是否结束。上面的例子可以看出,由于任务有2s的延时,在task1提交后立刻判断,task1还未完成,而在延时4s之后判断,task1就完成了。
    • 使用cancel()方法可以取消提交的任务,如果任务已经在线程池中运行了,就取消不了。这个例子中,线程池的大小设置为2,任务已经在运行了,所以取消失败。如果改变线程池的大小为1,那么先提交的是task1task2还在排队等候,这是时候就可以成功取消。
    • 使用result()方法可以获取任务的返回值。查看内部代码,发现这个方法是阻塞的。

    as_completed

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

    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
    
    executor = ThreadPoolExecutor(max_workers=2)
    urls = [3, 2, 4] # 并不是真的url
    all_task = [executor.submit(get_html, (url)) for 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

    as_completed()方法是一个生成器,在没有任务完成的时候,会阻塞,在有某个任务完成的时候,会yield这个任务,就能执行for循环下面的语句,然后继续阻塞住,循环到所有的任务结束。从结果也可以看出,先完成的任务会先通知主线程

    map

    除了上面的as_completed方法,还可以使用executor.map方法,但是有一点不同。

    from concurrent.futures import ThreadPoolExecutor
    import time
    
    # 参数times用来模拟网络请求的时间
    def get_html(times):
        time.sleep(times)
        print("get page {}s finished".format(times))
        return times
    
    executor = ThreadPoolExecutor(max_workers=2)
    urls = [3, 2, 4] # 并不是真的url
    
    for data in executor.map(get_html, urls):
        print("in main: get page {}s success".format(data))
    # 执行结果
    # 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

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

    wait

    wait方法可以让主线程阻塞,直到满足设定的要求。

    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
    
    executor = ThreadPoolExecutor(max_workers=2)
    urls = [3, 2, 4] # 并不是真的url
    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

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

    源码分析

    cocurrent.future模块中的future的意思是未来对象,可以把它理解为一个在未来完成的操作,这是异步编程的基础 。在线程池submit()之后,返回的就是这个future对象,返回的时候任务并没有完成,但会在将来完成。也可以称之为task的返回容器,这个里面会存储task的结果和状态。那ThreadPoolExecutor内部是如何操作这个对象的呢?

    下面简单介绍ThreadPoolExecutor的部分代码:

    1.init方法

    init方法中主要重要的就是任务队列和线程集合,在其他方法中需要使用到。

    ThreadPoolExecutor_init方法.png 

    2.submit方法

    submit中有两个重要的对象,_base.Future()_WorkItem()对象,_WorkItem()对象负责运行任务和对future对象进行设置,最后会将future对象返回,可以看到整个过程是立即返回的,没有阻塞。

    ThreadPoolExecutor_submit方法.png 

    3.adjust_thread_count方法

    这个方法的含义很好理解,主要是创建指定的线程数。但是实现上有点难以理解,比如线程执行函数中的weakref.ref,涉及到了弱引用等概念,留待以后理解。

     ThreadPoolExecutor_adjust_thread_count方法.png

     4._WorkItem对象

    _WorkItem对象的职责就是执行任务和设置结果。这里面主要复杂的还是self.future.set_result(result)

    _WorkItem类.png

    5.线程执行函数--_worker

    这是线程池创建线程时指定的函数入口,主要是从队列中依次取出task执行,但是函数的第一个参数还不是很明白。留待以后。

    ThreadPoolExecutor_线程执行函数_worker.png

    总结

    • future的设计理念很棒,在线程池/进程池和携程中都存在future对象,是异步编程的核心。
    • ThreadPoolExecutor 让线程的使用更加方便,减小了线程创建/销毁的资源损耗,无需考虑线程间的复杂同步,方便主线程与子线程的交互。
    • 线程池的抽象程度很高,多线程和多进程的编码接口一致。
    展开全文
  • 主要介绍了Python线程池模块ThreadPoolExecutor用法,结合实例形式分析了Python线程池模块ThreadPoolExecutor的导入与基本使用方法,需要的朋友可以参考下
  • 在开始发布这篇文章之前,我只是想补充一点,这是我第一次真正使用ThreadPoolExecutor,并且我感觉到我缺少明显的东西。我尝试查看现有帖子,但未找到任何相关内容,因此我将在此处发布查询。我在理解这部分内容时...

    在开始发布这篇文章之前,我只是想补充一点,这是我第一次真正使用ThreadPoolExecutor,并且我感觉到我缺少明显的东西。我尝试查看现有帖子,但未找到任何相关内容,因此我将在此处发布查询。我在理解这部分内容时遇到了一些麻烦,因此我将尽力做到尽可能简洁明了。

    我有以下代码片段:

    import time

    from queue import Queue

    from connection.api import IGAPI

    from concurrent.futures import ThreadPoolExecutor

    class MarketDataEngine:

    def __init__(self):

    self.market_data_stream = IGAPI()

    self.market_data_queue = Queue(maxsize=15)

    self.pool = ThreadPoolExecutor(2)

    def process_market_data(self, future):

    """

    Callback function to be applied on the future price object

    This is invoked once the future object has a status of done

    """

    if future.cancelled():

    print("{}: cancelled".format(future.arg))

    elif future.done():

    error = future.exception()

    if error:

    print("{}: error returned {}".format(future.arg, error))

    else:

    res = future.result()

    self.market_data_queue.put(res)

    #let queue know that thread is completed

    self.market_data_queue.task_done()

    def start(self):

    while True:

    #first sleep uncommented

    time.sleep(2)

    px = self.pool.submit(self.market_data_stream.execute, "get_market", "SPX 500")

    #second sleep commented out

    # time.sleep(2)

    self.pool.submit(self.process_market_data, px)

    # self.process_market_data(px)

    print("size of queue {}".format(self.market_data_queue.qsize()))

    if self.market_data_queue.full():

    print("Market data queue is full")

    removed_item = self.market_data_queue.get()

    print("item {} popped from queue".format(removed_item))

    if __name__ == "__main__":

    engine = MarketDataEngine()

    engine.start()

    我对以上问题的回答如下:

    1)当我运行上面的代码时,我得到以下输出:

    size of queue 0

    size of queue 0

    size of queue 0

    size of queue 0

    size of queue 0

    size of queue 0

    size of queue 0

    size of queue 0

    size of queue 0

    但是,如果我注释掉第一次睡眠而取消注释第二次睡眠,则会产生以下期望的输出:

    size of queue 1

    size of queue 1

    size of queue 2

    size of queue 3

    size of queue 4

    size of queue 5

    size of queue 6

    size of queue 7

    size of queue 8

    size of queue 9

    size of queue 10

    size of queue 11

    size of queue 12

    size of queue 13

    size of queue 14

    Market data queue is full

    item System Time: 2020-05-31 02:45:46.525632 Market Time: 22:02:21 Instrument: US 500 Bid: 3059.85 Ask: 3061.35 popped from queue

    size of queue 14

    Market data queue is full

    item System Time: 2020-05-31 02:45:48.511177 Market Time: 22:02:21 Instrument: US 500 Bid: 3059.85 Ask: 3061.35 popped from queue

    这种行为使我有些困惑,我无法弄清楚为什么会这样。

    如果我在调试器中以2xbreakpoints在以下行中用-Breakpoint表示了以下代码(未注释掉第一个睡眠,并注释了第二个睡眠):

    def start(self):

    while True:

    #first sleep uncommented

    time.sleep(2)

    *Breakpoint*px = self.pool.submit(self.market_data_stream.execute, "get_market", "SPX 500")

    #second sleep commented out

    # time.sleep(2)

    *Breakpoint*self.pool.submit(self.process_market_data, px)

    # self.process_market_data(px)

    print("size of queue {}".format(self.market_data_queue.qsize()))

    if self.market_data_queue.full():

    print("Market data queue is full")

    removed_item = self.market_data_queue.get()

    print("item {} popped from queue".format(removed_item))

    当我遍历代码序列时,可以获得所需的结果。我已经亲身意识到调试线程问题的难度,并且只是想知道是否有人可以简要解释为什么我遇到这个问题?如果有任何部分不完全清楚,我将非常乐于详细说明!

    展开全文
  • # -*- coding: utf-8 -*- import time from collections import OrderedDict from concurrent.futures import (ThreadPoolExecutor, as_completed) def get_thread_time(times): time.sleep(times) return ...
    # -*- coding: utf-8 -*-
    import time
    from concurrent.futures import ThreadPoolExecutor, as_completed
    
    def fib(n):
        if n < 3:
            return 1
        return fib(n-1) + fib(n-2)
    
    start_time = time.time()
    executor = ThreadPoolExecutor(max_workers=4)
    task_list = [executor.submit(fib, n) for n in range(3,35)]
    thread_results = [task.result() for task in as_completed(task_list)]
    print(thread_results)
    print("ThreadPoolExecutor time is: {}".format(time.time() - start_time))
    
    输出结果:
    [3, 8, 5, 2, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987, 1597, 2584, 
    4181, 6765, 10946, 28657, 17711, 46368, 75025, 121393, 
    196418, 317811, 514229, 832040, 1346269, 2178309, 3524578, 5702887]
    ThreadPoolExecutor time is: 1.7373127937316895
    

    总结

    1 future的设计理念很棒,在线程池/进程池和携程中都存在future对象,是异步编程的核心。
    2 ThreadPoolExecutor 让线程的使用更加方便,减小了线程创建/销毁的资源损耗,无需考虑线程间的复杂同步,方便主线程与子线程的交互。
    3 线程池的抽象程度很高,多线程和多进程的编码接口一致。

    展开全文
  • This is done to allow the interpreter # to exit when there are still idle threads in a ThreadPoolExecutor"s thread # pool (i.e. shutdown() was not called). However, allowing workers to die with # the...
  • APScheduler定时任务上次测试女神听了我的建议,已经做好了要给项目添加定时任务的决定了。但是之前提供的四种方式...Python中定时任务的解决方案,总体来说有四种,分别是:crontab、 scheduler、 Celery、 APSched...
  • python ThreadPoolExecutor 多线程踩坑之旅

    万次阅读 2019-08-10 15:18:58
    最近做了套接口测试系统,后台定时任务会批量去跑测试用例,用的是多线程,以前多线程模块用的是自己封装的包,但因为一些原因近期替换成了python标准库中的ThreadPoolExecutor,于是问题开始了…… 项目用的是:...
  • 这里主要想介绍 python `concurrent.futuresthread.ThreadPoolExecutor` 线程池中的 worker 引发异常的时候,并不会直接向上抛起异常,而是需要主线程通过调用`concurrent.futures.Future.exception(timeout=None)` ...
  • python有好几种实现线程池的方式,注意选择一下,哪种是自己需要的。最近用上ThreadPoolExecutor,用来代替celery的方式。主要是用在celery上的时间太少,没有吃透,还有,我们的应用没有复杂到那地步,多开几个...
  • 0 506 2019-12-06 13:09 − 在前面的文章中我们已经介绍了很多关于python线程相关的知识点,比如 线程互斥锁Lock / 线程事件Event / 线程条件变量Condition 等等,而今天给大家讲解的是&nb... 2019-12-24 10:31 − ...
  • Python ThreadPoolExecutor 异常打印

    千次阅读 2019-12-11 13:46:31
    利用ThreadPoolExecutor提交线程任务的时候,在线程里面发生异常,不会被主线程所捕抓,而是消无声息地死亡,非常不便于调试,尤其在我们写gui程序的时候。 我们可以新增一条线程帮我们把异常打印出来。 import time...
  • python有好几种实现线程池的方式,注意选择一下,哪种是自己需要的。最近用上ThreadPoolExecutor,用来代替celery的方式。主要是用在celery上的时间太少,没有吃透,还有,我们的应用没有复杂到那地步,多开几个...
  • 最近项目中由于需要频繁调用高德地图的接口,发现采用单线程循环调用太慢了,于是上网查了一下Python的并发,于是决定采用ThreadPoolExecutor(线程池)做并发。 在此之前,先看一下,单线程循环爬取多个网页的耗时...
  • 使用concurrent.futures并发执行简单任务用Python编写并发代码可能很棘手。在您开始之前,您必须考虑很多令人讨厌的事情,比如手头的任务是I/O密集型还是计算密集型、为实现并发性所付出的代价是否会给您带来所需要...
  • python ThreadPoolExecutor线程池(实例)

    千次阅读 2018-12-28 14:55:33
    Python中已经有了threading模块,为什么还需要线程池呢,线程池又是什么东西呢?在介绍线程同步的信号量机制的时候,举得例子是爬虫的例子,需要控制同时爬取的线程数,例子中创建了20个线程,而同时只允许3个线程在...
  • Python ThreadPoolExecutor 线程池导致内存暴涨 原文来源:https://www.cnblogs.com/xiaoweigege/p/15243893.html 背景 在工作任务中需要取抓取的时候,目前采用的是线程池去抓取,最终导致内存暴涨。 原因 ...
  • 1、如何使用线程池? 实际案例: 我们之前实现了一个多线程web...python3中有线程池实现,使用标准库中concurrent.futures下的ThreadPoolExecutor,对象的submit和map方法可以用来启动线程池中线程执行任务。 ...
  • python 并发 ThreadPoolExecutor 正文: Executor是一个抽象类,子类: ThreadPoolExecutor和ProcessPoolExecutor ,一个线程池,一个进程池. future对象:在未来的某一时刻完成操作的对象. submit方法可以返回一个...
  • Python3.2开始,标准库为我们提供了 concurrent.futures 模块,它提供了 ThreadPoolExecutor (线程池)和ProcessPoolExecutor (进程池)两个类。 相比 threading 等模块,该模块通过 submit 返回的是一个 future ...
  • 前言Python标准库为我们提供了threading和multiprocessing模块编写相应的多线程/多进程代码,但是当项目...但从Python3.2开始,标准库为我们提供了concurrent.futures模块,它提供了ThreadPoolExecutor和ProcessPool...
  • python3标准库concurrent.futures中常用的线程池ThreadPoolExecutor特点: 主线程可以获取某一个线程的状态,以及返回值。 线程同步 让多线程和多进程的编码接口一致。 简单粗暴 上手操练 将使用...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 12,296
精华内容 4,918
关键字:

pythonthreadpoolexecutor