《Python编程入门》12.2 concurrent.futures 与 multiprocessing

import time
from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch(name: str) -> str:
    time.sleep(0.1)
    return f"{name} done"


with ThreadPoolExecutor(max_workers=4) as ex:
    fut = ex.submit(fetch, "A")            # 提交后立刻返回 Future
    print("Future 类型:", type(fut).__name__)
    print("结果:", fut.result())           # result() 阻塞到完成

    print("map 保序:", list(ex.map(fetch, ["A", "B", "C", "D"])))

    futures = {ex.submit(fetch, n): n for n in ["A", "B", "C", "D"]}
    for f in as_completed(futures):        # 谁先完成谁先返回
        print("as_completed:", f.result())

Future 类型: Future
结果: A done
map 保序: ['A done', 'B done', 'C done', 'D done']
as_completed: B done
as_completed: D done
as_completed: A done
as_completed: C done

import time
from concurrent.futures import ThreadPoolExecutor


def boom(x: int) -> int:
    raise ValueError(f"处理 {x} 失败")


with ThreadPoolExecutor(max_workers=2) as ex:
    f = ex.submit(boom, 7)
    print("提交后主线程未崩溃，异常类型:", type(f.exception()).__name__)
    try:
        f.result()                         # 异常在这里才抛出
    except ValueError as e:
        print("调用 result() 时抛出:", e)

with ThreadPoolExecutor(max_workers=1) as ex:
    f = ex.submit(time.sleep, 1.0)
    try:
        f.result(timeout=0.2)
    except TimeoutError:
        print("超时: TimeoutError")

提交后主线程未崩溃，异常类型: ValueError
调用 result() 时抛出: 处理 7 失败
超时: TimeoutError

from concurrent.futures import ProcessPoolExecutor


def square(x: int) -> int:
    return x * x


with ProcessPoolExecutor(max_workers=2) as ex:   # 顶层代码，没有守卫
    print(list(ex.map(square, [1, 2, 3, 4])))

RuntimeError:
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

import os

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as ex:
        print(list(ex.map(square, [1, 2, 3, 4])))
        print("子进程 PID != 主进程 PID:",
              ex.submit(os.getpid).result(), "!=", os.getpid())

[1, 4, 9, 16]
子进程 PID != 主进程 PID: 54239 != 54211

ThreadPoolExecutor 首次任务(含建池): 0.0002s
ProcessPoolExecutor 首次任务(含建池): 0.2047s

200 个小任务 : 线程池 0.0009s | 进程池 0.0166s
4 个 CPU 任务: 线程池 2.166s | 进程池 0.555s

chunksize=   1: 0.5516s
chunksize= 100: 0.0034s
chunksize=1000: 0.0034s

import multiprocessing as mp


def producer(q, n):
    for i in range(n):
        q.put(i * i)


def accumulate(shared, n):
    for i in range(n):
        shared.append(i)


if __name__ == "__main__":
    q = mp.Queue()
    p = mp.Process(target=producer, args=(q, 5))
    p.start()
    p.join()
    print("Queue 收到:", [q.get() for _ in range(5)])

    with mp.Manager() as manager:
        shared = manager.list()
        procs = [mp.Process(target=accumulate, args=(shared, 3)) for _ in range(2)]
        for pr in procs:
            pr.start()
        for pr in procs:
            pr.join()
        print("Manager.list 长度:", len(shared), "内容:", list(shared))

    plain = []
    pr = mp.Process(target=accumulate, args=(plain, 3))
    pr.start()
    pr.join()
    print("普通 list 长度:", len(plain))

Queue 收到: [0, 1, 4, 9, 16]
Manager.list 长度: 6 内容: [0, 1, 2, 0, 1, 2]
普通 list 长度: 0

import os
print("os.cpu_count():", os.cpu_count())
print("os.process_cpu_count():", os.process_cpu_count())

os.cpu_count(): 10
os.process_cpu_count(): 10