Python 并发调用实战

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from concurrent.futures import ThreadPoolExecutor
from time import sleep


def fetch_price(product_id, *, currency="CNY"):
    sleep(0.05)
    return {"product_id": product_id, "price": product_id * 10, "currency": currency}


product_ids = [101, 102, 103]
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(fetch_price, product_id, currency="CNY") for product_id in product_ids]
    results = [future.result() for future in futures]

for result in results:
    print(result)

# 输出：
# {'product_id': 101, 'price': 1010, 'currency': 'CNY'}
# {'product_id': 102, 'price': 1020, 'currency': 'CNY'}
# {'product_id': 103, 'price': 1030, 'currency': 'CNY'}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from concurrent.futures import ThreadPoolExecutor, as_completed
from time import sleep


def fetch_price(product_id, *, currency="CNY"):
    sleep(0.05)
    return {"product_id": product_id, "price": product_id * 10, "currency": currency}


product_ids = [101, 102, 103]
results = [None] * len(product_ids)
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(fetch_price, item): index for index, item in enumerate(product_ids)}
    for future in as_completed(futures):
        index = futures[future]
        results[index] = future.result()
        # 在这里更新进度或保存该项结果。

print([item["price"] for item in results])

# 输出：[1010, 1020, 1030]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

from concurrent.futures import ThreadPoolExecutor, as_completed
from functools import partial
from time import perf_counter, sleep
from loguru import logger


def run_calls(calls, max_workers=8):
    started = perf_counter()
    results, errors = {}, {}

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(call): name for name, call in calls.items()}
        for future in as_completed(futures):
            name = futures[future]
            try:
                results[name] = future.result()
            except Exception as exc:
                errors[name] = exc
                logger.exception(f"任务失败：{name}")
            else:
                logger.info(f"任务成功：{name}")

    logger.info(f"任务数={len(calls)}，成功={len(results)}，失败={len(errors)}，总耗时={perf_counter() - started:.3f} 秒")
    return results, errors


def fetch_profile(user_id):
    sleep(0.05)
    return {"user_id": user_id, "name": "Lin"}


def fetch_orders(user_id, *, limit=2):
    sleep(0.03)
    return [user_id * 100 + index for index in range(limit)]


def fetch_coupon(user_id):
    raise ConnectionError("优惠服务不可用（模拟）")


calls = {
    "profile": partial(fetch_profile, 7),
    "orders": partial(fetch_orders, 7, limit=2),
    "coupon": partial(fetch_coupon, 7),
}
results, errors = run_calls(calls, max_workers=3)
print(results["profile"])
print(results["orders"])
print(sorted(errors))

# print 输出如下；日志顺序和耗时不固定。
# {'user_id': 7, 'name': 'Lin'}
# [700, 701]
# ['coupon']

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

import json
from collections import Counter
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import get_context
from pathlib import Path
from tempfile import TemporaryDirectory


def count_errors(path):
    counts = Counter()
    with path.open(encoding="utf-8") as source:
        for line in source:
            record = json.loads(line)
            if 500 <= record["status"] < 600:
                counts[record["service"]] += 1
    return counts


def main():
    batches = [
        [{"service": "orders", "status": 200}, {"service": "payments", "status": 502}],
        [{"service": "orders", "status": 500}, {"service": "payments", "status": 503}],
    ]
    with TemporaryDirectory() as directory:
        paths = []
        for index, records in enumerate(batches):
            path = Path(directory) / f"requests-{index}.jsonl"
            path.write_text("".join(json.dumps(record) + "\n" for record in records), encoding="utf-8")
            paths.append(path)

        totals = Counter()
        context = get_context("spawn")
        with ProcessPoolExecutor(max_workers=2, mp_context=context) as executor:
            for counts in executor.map(count_errors, paths):
                totals.update(counts)

    print(dict(sorted(totals.items())))


if __name__ == "__main__":
    main()

# 输出：{'orders': 1, 'payments': 2}