Python 异步编程与协程详解

Python 异步编程与协程详解 | 极客日志

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 运行协程
asyncio.run(hello())

import asyncio

# 定义协程
async def count():
    print('Counting...')
    await asyncio.sleep(1)
    print('Counted!')

# 运行多个协程
async def main():
    await asyncio.gather(count(), count(), count())

asyncio.run(main())

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 创建任务
async def main():
    task1 = asyncio.create_task(hello())
    task2 = asyncio.create_task(hello())
    await task1
    await task2

asyncio.run(main())

import asyncio

# 定义协程
async def hello():
    try:
        print('Hello, World!')
        await asyncio.sleep(1)
        print('Hello again!')
    except asyncio.CancelledError:
        print('Task cancelled!')

# 创建任务并取消
async def main():
    task = asyncio.create_task(hello())
    await asyncio.sleep(0.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('Main: Task cancelled!')

asyncio.run(main())

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(2)
    print('Hello again!')

# 任务超时
async def main():
    try:
        await asyncio.wait_for(hello(), timeout=1)
    except asyncio.TimeoutError:
        print('Task timed out!')

asyncio.run(main())

import asyncio

# 获取事件循环
loop = asyncio.get_event_loop()

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 运行协程
loop.run_until_complete(hello())

import asyncio

# 获取事件循环
loop = asyncio.get_event_loop()

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 运行多个协程
loop.run_until_complete(asyncio.gather(hello(), hello(), hello()))

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 运行协程
asyncio.run(hello())

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 使用 Future 对象
async def main():
    future = asyncio.Future()
    task = asyncio.create_task(hello())
    task.add_done_callback(lambda t: future.set_result(t.result()))
    await future

asyncio.run(main())

pip install aiohttp

import aiohttp
import asyncio

# 发送 GET 请求
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)

asyncio.run(main())

import aiohttp
import asyncio
import json

# 发送 POST 请求
async def post_data(session, url, data):
    async with session.post(url, data=data) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = {'name': '张三', 'age': 25}
        response = await post_data(session, 'https://httpbin.org/post', data)
        print(response)

asyncio.run(main())

import aiohttp
import asyncio
import json

# 发送 JSON 请求
async def post_json(session, url, data):
    async with session.post(url, json=data) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = {'name': '张三', 'age': 25}
        response = await post_json(session, 'https://httpbin.org/post', data)
        print(response)

asyncio.run(main())

import aiohttp
import asyncio
import time

# 发送 HTTP 请求
async def fetch(session, url):
    start_time = time.time()
    async with session.get(url) as response:
        text = await response.text()
        elapsed_time = time.time() - start_time
        return url, len(text), elapsed_time

# 并发发送 HTTP 请求
async def main():
    urls = [
        'https://www.example.com',
        'https://www.google.com',
        'https://www.github.com',
        'https://www.python.org',
        'https://www.djangoproject.com'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, length, elapsed_time in results:
            print(f'URL: {url}, 响应长度：{length}, 耗时：{elapsed_time:.2f}秒')

# 运行程序
if __name__ == '__main__':
    start_time = time.time()
    asyncio.run(main())
    elapsed_time = time.time() - start_time
    print(f'总耗时：{elapsed_time:.2f}秒')

from aiohttp import web
import asyncio

# 处理根路径请求
async def handle_root(request):
    return web.Response(text='Hello, World!')

# 处理 API 接口请求
async def handle_api(request):
    data = {'name': '张三', 'age': 25}
    return web.json_response(data)

# 提供静态文件服务
async def handle_static(request):
    return web.FileResponse('static/index.html')

# 创建 Web 应用
async def create_app():
    app = web.Application()
    # 添加路由
    app.add_routes([
        web.get('/', handle_root),
        web.get('/api', handle_api),
        web.get('/static/{name}', handle_static)
    ])
    return app

# 运行 Web 服务器
if __name__ == '__main__':
    asyncio.run(web.run_app(create_app(), host='localhost', port=8080))

Python 异步编程与协程详解

异步编程与协程

学习目标与重点

异步编程概述

什么是异步编程

异步编程的优势

异步编程的应用场景

协程的定义与使用

协程的定义

协程的使用

协程的暂停与恢复

任务调度

创建任务

任务的取消

任务的超时

事件循环

事件循环的概述

获取事件循环

事件循环的运行

asyncio 库

asyncio 的基本用法

asyncio 的常用函数

asyncio 的高级用法

aiohttp 库

安装 aiohttp

发送 HTTP 请求

发送 POST 请求

发送 JSON 请求

实战案例：异步 HTTP 客户端

需求分析

代码实现

实施过程

最终效果

实战案例：异步 Web 服务器

需求分析

代码实现

实施过程

最终效果

总结

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具