Python协程入门

 1async def delay_add(x: int, y: int, delay: int = 1) -> int:
 2    print(f"adding {x} and {y}")
 3    return x + y
 4
 5
 6print(type(delay_add))
 7
 8coro = delay_add(1, 2)
 9
10print(type(coro))

1<class 'function'>
2<class 'coroutine'>
3<sys>:0: RuntimeWarning: coroutine 'delay_add' was never awaited

 1import asyncio
 2
 3
 4async def delay_add(x: int, y: int, delay: int = 1) -> int:
 5    print(f"adding {x} and {y}")
 6    return x + y
 7
 8
 9loop = asyncio.new_event_loop()
10task = loop.create_task(delay_add(1, 2))
11loop.run_until_complete(task)
12print(task.result())
13loop.close()

 1import asyncio
 2import time
 3
 4BASE_TIME = time.time()
 5
 6
 7async def delay_calc(x: int, y: int, delay: int = 1) -> int:
 8    await asyncio.sleep(delay)
 9    return x + y
10
11
12# 演示一下await赋值
13async def wrapper(x: int, y: int, delay: int = 1) -> None:
14    print(f"adding {x} and {y}, time: {time.time() - BASE_TIME}")
15    res = await delay_calc(x, y, delay)
16    print(f"result of {x} + {y}: {res}, time: {time.time() - BASE_TIME}")
17
18
19if __name__ == "__main__":
20    loop = asyncio.new_event_loop()
21    loop.run_until_complete(
22        asyncio.gather(
23            loop.create_task(wrapper(1, 2)),
24            loop.create_task(wrapper(3, 4)),
25        )
26    )
27    loop.close()

1adding 1 and 2, time: 0.0007719993591308594
2adding 3 and 4, time: 0.0009663105010986328
3result of 1 + 2: 3, time: 1.0045411586761475
4result of 3 + 4: 7, time: 1.0047202110290527

 1import asyncio
 2import time
 3
 4import httpx
 5
 6ASYNC_HTTPX_CLIENT = httpx.AsyncClient()
 7
 8
 9async def _get_index_length_single(url: str) -> int:
10    try:
11        response = await ASYNC_HTTPX_CLIENT.get(url)
12        return len(response.text)
13    except Exception:
14        return -1
15
16
17async def get_index_length(urls: list[str]) -> list[int]:
18    time_start = time.time()
19    results = await asyncio.gather(*(_get_index_length_single(url) for url in urls))
20    time_end = time.time()
21    print(f"URLs: {len(urls)}, Time: {time_end - time_start}")
22    return results
23
24
25async def main():
26    urls = [
27        "https://www.bing.com",
28        "https://www.baidu.com",
29        "https://www.sogou.com",
30        "https://www.zhihu.com",
31        "https://www.bilibili.com",
32    ]
33    print(await get_index_length(urls[:1]))
34    print(await get_index_length(urls[1:]))
35
36
37if __name__ == "__main__":
38    asyncio.run(main())

 1import asyncio
 2from random import randint
 3from typing import AsyncGenerator
 4
 5
 6async def content_generator() -> AsyncGenerator[str, None]:
 7    text = "你好，我是一个大模型应用。"
 8    i, j = 0, len(text)
 9    while i < j:
10        delta = randint(1, 5)
11        yield text[i : i + delta]
12        i += delta
13        await asyncio.sleep(0.1)
14
15
16async def sse_generator() -> AsyncGenerator[str, None]:
17    async for content in content_generator():
18        yield f'data: {{"content":"{content}"}}\n\n'
19
20
21async def client():
22    async for line in sse_generator():
23        print(line)
24
25
26if __name__ == "__main__":
27    asyncio.run(client())

 1import asyncio
 2import random
 3import time
 4
 5
 6async def random_task() -> float:
 7    sleep_time = random.random() * 5
 8    await asyncio.sleep(sleep_time)
 9    return sleep_time
10
11
12async def main():
13    pending_set = [asyncio.create_task(random_task()) for _ in range(10)]
14    done_set = None
15
16    results = []
17
18    start_time = time.time()
19
20    while pending_set:
21        done_set, pending_set = await asyncio.wait(pending_set, timeout=1.0)
22        cost_time = time.time() - start_time
23
24        for task in done_set:
25            results.append(task.result())
26
27        print(
28            "cost time: {:.2f}, pending: {}, done: {}".format(
29                cost_time, len(pending_set), len(results)
30            )
31        )
32
33        if cost_time > 3:
34            print("超时，未完成任务将被取消")
35            for task in pending_set:
36                task.cancel()
37            break
38
39    print(results)
40
41
42if __name__ == "__main__":
43    asyncio.run(main())

 1import asyncio
 2import time
 3
 4BASE_TIME = time.time()
 5
 6
 7async def delay_calc(x: int, y: int, delay: int = 1) -> None:
 8    print(f"adding {x} and {y}, time: {time.time() - BASE_TIME}")
 9    await asyncio.to_thread(time.sleep, delay)
10    print(f"result of {x} + {y}: {x + y}, time: {time.time() - BASE_TIME}")
11
12
13if __name__ == "__main__":
14    loop = asyncio.new_event_loop()
15    loop.run_until_complete(
16        asyncio.gather(
17            loop.create_task(delay_calc(1, 2)),
18            loop.create_task(delay_calc(3, 4)),
19        )
20    )
21    loop.close()

 1import asyncio
 2import math
 3import random
 4import time
 5from concurrent.futures import ProcessPoolExecutor
 6
 7
 8def cpu_bound_calc(task_name: str) -> float:
 9    start_time = time.time()
10    length = 10**7
11    res = sum(math.sqrt(random.random()) for _ in range(length)) / length
12    print(f"{task_name} cost time: {time.time() - start_time}")
13    return res
14
15
16async def main():
17    loop = asyncio.get_running_loop()
18    start_time = time.time()
19    with ProcessPoolExecutor(max_workers=2) as executor:
20        results = await asyncio.gather(
21            loop.run_in_executor(executor, cpu_bound_calc, "task1"),
22            loop.run_in_executor(executor, cpu_bound_calc, "task2"),
23        )
24    print(f"cost time: {time.time() - start_time}")
25    print(results)
26
27
28if __name__ == "__main__":
29    asyncio.run(main())

1task1 cost time: 1.3992047309875488
2task2 cost time: 1.3943092823028564
3cost time: 1.588423728942871
4[0.6665390533899224, 0.6666580988367764]

 1import asyncio
 2import math
 3import random
 4import time
 5
 6
 7def cpu_bound_calc(task_name: str) -> float:
 8    start_time = time.time()
 9    length = 10**7
10    res = sum(math.sqrt(random.random()) for _ in range(length)) / length
11    print(f"{task_name} cost time: {time.time() - start_time}")
12    return res
13
14
15async def main():
16    start_time = time.time()
17    results = await asyncio.gather(
18        asyncio.to_thread(cpu_bound_calc, "task1"),
19        asyncio.to_thread(cpu_bound_calc, "task2"),
20    )
21    print(f"cost time: {time.time() - start_time}")
22    print(results)
23
24
25if __name__ == "__main__":
26    asyncio.run(main())

 1import asyncio
 2import time
 3
 4import httpx
 5
 6ASYNC_HTTPX_CLIENT = httpx.AsyncClient()
 7
 8
 9async def _get_index_length_single(url: str) -> int:
10    return len((await ASYNC_HTTPX_CLIENT.get(url)).text)
11
12
13async def get_index_length(urls: list[str]) -> list[int]:
14    time_start = time.time()
15
16    tasks = []
17    results = []
18
19    try:
20        async with asyncio.TaskGroup() as tg:
21            for url in urls:
22                tasks.append(tg.create_task(_get_index_length_single(url)))
23        results = [task.result() for task in tasks]
24    except ExceptionGroup as eg:
25        print(f"捕获到异常组:{eg}")
26        for e in eg.exceptions:
27            print(f"具体异常:{e}")
28
29    time_end = time.time()
30    print(f"URLs: {len(urls)}, Time: {time_end - time_start}")
31
32    return results
33
34
35async def main():
36    urls = [
37        "https://www.bing.com",
38        "https://www.baidu.com",
39        "https://www.sogou.com",
40        "https://www.zhihu.com",
41        "https://www.bilibili.com",
42    ]
43    print(await get_index_length(urls[:1] + ["https://bad.url"] * 2))
44    print(await get_index_length(urls[1:]))
45
46
47if __name__ == "__main__":
48    asyncio.run(main())

1捕获到异常组:unhandled errors in a TaskGroup (2 sub-exceptions)
2具体异常:[Errno 11001] getaddrinfo failed
3具体异常:[Errno 11001] getaddrinfo failed
4URLs: 3, Time: 0.01521611213684082
5[]
6URLs: 4, Time: 0.13132429122924805
7[28918, 19011, 118, 3286]