Python 异步编程进阶与 asyncio 高级应用

任务管理

import asyncio

async def task1():
    print("Task 1 start")
    await asyncio.sleep(1)
    print("Task 1 end")

async def task2():
    print("Task 2 start")
    await asyncio.sleep(2)
    print("Task 2 end")

async def main():
    print("Main start")
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    await task1_obj
    await task2_obj
    print("Main end")

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

异步 IO 操作

import asyncio
import aiohttp

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

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, "https://www.example.com")
        print(html)

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

并发控制

import asyncio
import aiohttp
from asyncio import Semaphore

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

async def main():
    urls = ["https://www.example.com"] * 10
    sem = Semaphore(3)
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url, sem)) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(len(result))

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

实战案例：异步爬虫

需求分析

开发一个异步爬虫，支持以下功能：

• 异步爬取多个网页。
• 解析网页内容。
• 保存爬取结果。
• 并发控制。

代码实现

import asyncio
import aiohttp
from asyncio import Semaphore
from bs4 import BeautifulSoup

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

async def parse(html):
    soup = BeautifulSoup(html, "html.parser")
    title = soup.find("title").text
    return title

async def save_result(title, url):
    with open("results.txt", "a") as f:
        f.write(f"{title}\t{url}\n")

async def crawl(url, session, sem):
    try:
        html = await fetch(session, url, sem)
        title = await parse(html)
        await save_result(title, url)
        print(f"Successfully crawled {url}")
    except Exception as e:
        print(f"Failed to crawl {url}: {e}")

async def main():
    urls = ["https://www.example.com", "https://www.google.com", "https://www.github.com", "https://www.python.org", "https://www.djangoproject.com"]
    sem = Semaphore(3)
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(crawl(url, session, sem)) for url in urls]
        await asyncio.gather(*tasks)

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

实施过程

1. 安装依赖：aiohttp、beautifulsoup4。
2. 实现异步爬取和解析网页内容。
3. 实现保存爬取结果。
4. 实现并发控制。
5. 运行应用。

最终效果

通过异步爬虫，可以实现以下功能：

• 异步爬取多个网页。
• 解析网页内容。
• 保存爬取结果。
• 并发控制。

实战案例：异步 Web 服务器

需求分析

开发一个异步 Web 服务器，支持以下功能：

• 处理 HTTP 请求。
• 异步读取文件。
• 异步数据库操作。
• 并发控制。

代码实现

import asyncio
from aiohttp import web
import aiofiles
import sqlite3
from asyncio import Semaphore

async def read_file(file_path):
    async with aiofiles.open(file_path, "r") as f:
        return await f.read()

async def get_data_from_db():
    conn = sqlite3.connect("test.db")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    data = cursor.fetchall()
    conn.close()
    return data

async def index(request):
    content = await read_file("index.html")
    return web.Response(text=content, content_type="text/html")

async def users(request):
    sem = Semaphore(3)
    async with sem:
        data = await get_data_from_db()
        return web.json_response(data)

async def create_app():
    app = web.Application()
    app.add_routes([
        web.get("/", index),
        web.get("/users", users)
    ])
    return app

if __name__ == "__main__":
    asyncio.run(web.run_app(create_app(), host="0.0.0.0", port=8000))

实施过程

1. 安装依赖：aiohttp、aiofiles。
2. 实现异步读取文件和数据库操作。
3. 实现 HTTP 请求处理。
4. 实现并发控制。
5. 运行应用。

最终效果

通过异步 Web 服务器，可以实现以下功能：

• 处理 HTTP 请求。
• 异步读取文件。
• 异步数据库操作。
• 并发控制。

总结

本文详细介绍了 Python 异步编程的高级技巧，包括协程调度、任务管理、异步 IO 操作、并发控制等；学习了 asyncio 库的高级应用；通过实战案例开发了异步爬虫和异步 Web 服务器。