Python算法
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