ThreadPoolExecutor

لماذا نستخدم تجمع خيوط؟

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor() as ex:
    future = ex.submit(square, 5)
    print(future.result())

submit وFuture

from concurrent.futures import ThreadPoolExecutor

def greet(name):
    return 'hi ' + name

with ThreadPoolExecutor(max_workers=2) as ex:
    f1 = ex.submit(greet, 'Ana')
    f2 = ex.submit(greet, 'Bob')
    print(f1.result())
    print(f2.result())

map لمدخلات متعددة

from concurrent.futures import ThreadPoolExecutor

def cube(n):
    return n ** 3

with ThreadPoolExecutor() as ex:
    results = ex.map(cube, range(5))
    print(list(results))

التحكم في عدد خيوط العمل

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(n):
    time.sleep(0.05)
    return n

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as ex:
    print(list(ex.map(io_task, range(4))))
print('ran concurrently')

as_completed

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def work(n):
    time.sleep(0.01 * (3 - n))
    return n

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(work, i) for i in range(3)]
    for f in as_completed(futures):
        print('finished', f.result())

معالجة الاستثناءات

from concurrent.futures import ThreadPoolExecutor

def risky(n):
    if n == 0:
        raise ValueError('cannot be zero')
    return 10 // n

with ThreadPoolExecutor() as ex:
    f = ex.submit(risky, 0)
    try:
        print(f.result())
    except ValueError as e:
        print('caught:', e)

التعيين باستخدام عدة وسيطات

from concurrent.futures import ThreadPoolExecutor

def add(a, b):
    return a + b

with ThreadPoolExecutor() as ex:
    print(list(ex.map(add, [1, 2, 3], [10, 20, 30])))

جمع النتائج في قاموس

from concurrent.futures import ThreadPoolExecutor, as_completed

def length(word):
    return len(word)

words = ['cat', 'tiger', 'ox']
with ThreadPoolExecutor() as ex:
    future_to_word = {ex.submit(length, w): w for w in words}
    out = {}
    for f in as_completed(future_to_word):
        out[future_to_word[f]] = f.result()
print(sorted(out.items()))

إيقاف التشغيل باستخدام مدير السياق

from concurrent.futures import ThreadPoolExecutor

def job(n):
    return n * 2

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(job, i) for i in range(3)]
print('all tasks done after the with block')
print([f.result() for f in futures])

التحقق من حالة Future

from concurrent.futures import ThreadPoolExecutor

def quick(n):
    return n + 1

with ThreadPoolExecutor() as ex:
    f = ex.submit(quick, 41)
    result = f.result()
    print('done?', f.done())
    print('value:', result)

متى تستخدمونه؟

from concurrent.futures import ThreadPoolExecutor

urls = ['a', 'b', 'c']

def fetch(u):
    return 'fetched ' + u

with ThreadPoolExecutor(max_workers=3) as ex:
    for r in ex.map(fetch, urls):
        print(r)

اختبار سريع

مراجعة