时间戳

首页 > 解决方案 > 从 concurrent.futures 获取进度更新

python-3.x - 从 concurrent.futures 获取进度更新

问题描述

我想从单独的线程或进程（以更快者为准）复制文件，以免阻塞主线程。

我还想偶尔获得进度更新。

使用“常规”工作线程，我可以将进度统计信息推送到 a Queue，并从主（UI）线程检查它。

我该怎么做concurrent.futures呢？

标签： python-3.xconcurrencyconcurrent.futures

解决方案

替代方法

background_runner用于运行后台线程/任务的装饰器/注释。background_pools用于保存当前正在运行的线程及其进度。__context掌握进展。

import threading
from collections import defaultdict
import time

background_pools = defaultdict(lambda: {})

def background_runner(_func=None, *, pool_max=1, pool_name='default'):
    def main_wrapper(task): # It is our internal decorator, and (task) is our decorated function.
        pool_size=pool_max
        global background_pools

        # It will return empty array if pool is not found.
        pool = background_pools[pool_name]

        print("Pool name is:",pool_name)
        print("Pool size is:",pool_size)

        def task_wrapper(*args, **kwargs): # It is the replacement or Decorated version of aur (task) or (_func)
            def task_in_thread():
                thread_id = threading.current_thread().ident
                context = {}
                pool[thread_id] = { "thread": threading.current_thread(), "context":context}
                try:
                    return task(*args, **kwargs, __context=context)
                finally:
                    try: 
                        del pool[thread_id]
                    except:
                        pass    

            if len(pool.keys()) < pool_size:
                threading.Thread(target=task_in_thread).start()
                print("Task:'{}' is in process.".format(pool_name))
            else:
                print(f"Only { pool_size } task:{pool_name} can run at a time.")
        return task_wrapper
    if _func is None:
        # decorator is used with named arguments.
        return main_wrapper                
    else:
        # decorator is used without arguments.
        return main_wrapper(_func)

使用. background_runner_ 用于更新进度。time.sleep__context

@background_runner(pool_max=3, pool_name='sleep_test')
def sleep_test(__context={}):
    __context['time'] = 0
    for index in range(0, 20):
        time.sleep(2)
        __context['time'] += 2

测试方法的调用

sleep_test()
time.sleep(10) 
print(background_pools)
sleep_test()
time.sleep(10)
print(background_pools)
time.sleep(10)
sleep_test()
sleep_test()
print(background_pools)
time.sleep(10)
print(background_pools)

推荐阅读