Python异步编程入门教程指南

# 同步代码
import time

def sync_task(name):
    print(f"Start {name}")
    time.sleep(3)
    print(f"End {name}")

sync_task("Task 1")
sync_task("Task 2")

# 异步代码
import asyncio

async def async_task(name):
    print(f"Start {name}")
    await asyncio.sleep(3)
    print(f"End {name}")

async def main():
    await asyncio.gather(async_task("Task 1"), async_task("Task 2"))

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

import asyncio

async def async_example():
    print("Start")
    await asyncio.sleep(2)
    print("End")

if __name__ == "__main__":
    asyncio.run(async_example())

import aiohttp
import asyncio

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    url = "https://jsonplaceholder.typicode.com/posts/1"
    result = await fetch_data(url)
    print(result)

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

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com", "https://example.org", "https://example.net"]

    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)

    for result in results:
        print(result)

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

import asyncio

async def task1():
    print("Task 1")
    await asyncio.sleep(2)
    print("Task 1 completed")

async def task2():
    print("Task 2")
    await asyncio.sleep(1)
    print("Task 2 completed")

async def main():
    await asyncio.gather(task1(), task2())

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

import asyncio

async def async_example():
    try:
        # 异步代码
        await asyncio.sleep(2)
        raise ValueError("An async error")
    except ValueError as e:
        print(f"Caught an error: {e}")

if __name__ == "__main__":
    asyncio.run(async_example())

import asyncio

async def task1(lock1, lock2):
    async with lock1:
        print("Task 1 acquired lock1")
        await asyncio.sleep(1)
        print("Task 1 waiting for lock2")
        async with lock2:
            print("Task 1 acquired lock2")

async def task2(lock1, lock2):
    async with lock2:
        print("Task 2 acquired lock2")
        await asyncio.sleep(1)
        print("Task 2 waiting for lock1")
        async with lock1:
            print("Task 2 acquired lock1")

async def main():
    lock1 = asyncio.Lock()
    lock2 = asyncio.Lock()

    await asyncio.gather(task1(lock1, lock2), task2(lock1, lock2))

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