Python 异步编程进阶与 asyncio 高级应用
Python 异步编程的高级技巧,包括协程调度、任务管理、异步 IO 操作及并发控制。通过 asyncio 库的应用示例和实战案例(异步爬虫、Web 服务器),展示了如何构建高性能并发程序,旨在提升应用程序的吞吐量和响应速度。

Python 异步编程的高级技巧,包括协程调度、任务管理、异步 IO 操作及并发控制。通过 asyncio 库的应用示例和实战案例(异步爬虫、Web 服务器),展示了如何构建高性能并发程序,旨在提升应用程序的吞吐量和响应速度。

学习目标:掌握 Python 异步编程的高级技巧,包括协程调度、任务管理、异步 IO 操作、并发控制等;学习 asyncio 库的高级应用;通过实战案例开发异步应用程序。
学习重点:协程调度、任务管理、异步 IO 操作、并发控制、asyncio 高级应用、异步应用实战。
异步编程是一种编程模型,通过非阻塞的方式处理 IO 操作,提高应用程序的并发性能。在异步编程中,任务可以在等待 IO 操作完成时继续执行其他任务,而不需要阻塞等待。
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")
await asyncio.gather(task1(), task2())
print("Main end")
if __name__ == "__main__":
asyncio.run(main())
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())
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())
通过异步爬虫,可以实现以下功能:
开发一个异步 Web 服务器,支持以下功能:
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))
通过异步 Web 服务器,可以实现以下功能:
本文详细介绍了 Python 异步编程的高级技巧,包括协程调度、任务管理、异步 IO 操作、并发控制等;学习了 asyncio 库的高级应用;通过实战案例开发了异步爬虫和异步 Web 服务器。

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online
将 Markdown(GFM)转为 HTML 片段,浏览器内 marked 解析;与 HTML转Markdown 互为补充。 在线工具,Markdown转HTML在线工具,online
将 HTML 片段转为 GitHub Flavored Markdown,支持标题、列表、链接、代码块与表格等;浏览器内处理,可链接预填。 在线工具,HTML转Markdown在线工具,online