异步、并发与超时：等待的时候，程序还能做什么

py -3.13 -m venv .venv
$env:Path = "$PWD\.venv\Scripts;$env:Path"
python -c "import sys; print(sys.executable)"

# 按固定输入观察异步、并发与超时的处理过程和返回结果。
import asyncio
from time import perf_counter

async def read_one(name, delay):
    await asyncio.sleep(delay)
    return name

async def main():
    started = perf_counter()
    first = await read_one("A", 0.12)
    second = await read_one("B", 0.04)
    print([first, second])
    print(f"耗时约 {perf_counter() - started:.2f} 秒")

asyncio.run(main())

# 按固定输入观察异步、并发与超时的处理过程和返回结果。
import asyncio
from time import perf_counter

async def read_one(name, delay):
    await asyncio.sleep(delay)
    return name

async def main():
    started = perf_counter()
    results = await asyncio.gather(
        read_one("A", 0.12),
        read_one("B", 0.04),
    )
    print(results)
    print(f"耗时约 {perf_counter() - started:.2f} 秒")

asyncio.run(main())

# 按固定输入观察异步、并发与超时的处理过程和返回结果。
import asyncio

async def read_one(name, delay, slots):
    async with slots:
        print(name, "开始")
        await asyncio.sleep(delay)
        print(name, "结束")

async def main():
    slots = asyncio.Semaphore(2)
    await asyncio.gather(
        read_one("A", 0.12, slots),
        read_one("B", 0.04, slots),
        read_one("C", 0.08, slots),
    )

asyncio.run(main())

"""比较依次等待、同时等待和超时三种任务执行方式。"""
import asyncio
import json
import time

async def run_batch(items, limit, timeout):
    if type(limit) is not int or limit < 1 or timeout <= 0:
        raise ValueError('positive limit and timeout required')
    # limit 是同时运行的任务上限；拿到一个名额后，任务才能开始。
    semaphore = asyncio.Semaphore(limit)
    active = peak = 0
    completion_order = []
    async def job(name, delay):
        nonlocal active, peak
        # 先等运行名额，再开始计时；排队时间在这段超时计时之外。
        async with semaphore:
            active += 1
            peak = max(peak, active)
            try:
                async with asyncio.timeout(timeout):
                    await asyncio.sleep(delay)
                completion_order.append(name)
                return {'name': name, 'status': 'done'}
            except TimeoutError:
                return {'name': name, 'status': 'timeout'}
            finally:
                active -= 1
    start = time.perf_counter()
    # gather 一起安排这些任务，并按传入顺序收集结果；完成顺序另外记录。
    results = await asyncio.gather(*(job(name, delay) for name, delay in items))
    return {'results': results, 'completion_order': completion_order, 'peak': peak, 'active_after': active, 'elapsed_seconds': round(time.perf_counter() - start, 4)}

async def main():
    items = [('A', .12), ('B', .04), ('C', .08)]
    for label, limit, timeout in [('serial', 1, 1), ('concurrent', 2, 1), ('timeout', 2, .06)]:
        print(label, json.dumps(await run_batch(items, limit, timeout), ensure_ascii=False))

if __name__ == '__main__':
    asyncio.run(main())

python -X utf8 async_lab.py

import asyncio
from async_lab import run_batch

result = asyncio.run(run_batch([("资料A", 0.11)], 1, 0.10))
print(result)

serial {"results": [{"name": "A", "status": "done"}, {"name": "B", "status": "done"}, {"name": "C", "status": "done"}], "completion_order": ["A", "B", "C"], "peak": 1, "active_after": 0, "elapsed_seconds": 0.262}
concurrent {"results": [{"name": "A", "status": "done"}, {"name": "B", "status": "done"}, {"name": "C", "status": "done"}], "completion_order": ["B", "A", "C"], "peak": 2, "active_after": 0, "elapsed_seconds": 0.1411}
timeout {"results": [{"name": "A", "status": "timeout"}, {"name": "B", "status": "done"}, {"name": "C", "status": "timeout"}], "completion_order": ["B"], "peak": 2, "active_after": 0, "elapsed_seconds": 0.1087}

python -X utf8 -m unittest test_async_lab -v