Python 异步网络编程概览

asyncio 实现协程异步 IO，高效处理大量并发网络连接。

asyncio 基本概念

import asyncio

# 定义协程
async def hello():
    print("Hello")
    await asyncio.sleep(1)  # 异步等待
    print("World")

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

协程并发

import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    print(f"任务 {name} 完成")

async def main():
    # 并发执行多个协程
    await asyncio.gather(
        task('A', 1),
        task('B', 2),
        task('C', 1)
    )

asyncio.run(main())
# 输出顺序: A, C, B（按完成时间）

asyncio TCP 客户端

import asyncio

async def tcp_client():
    reader, writer = await asyncio.open_connection('localhost', 8080)

    # 发送数据
    writer.write(b'Hello\n')
    await writer.drain()

    # 接收数据
    data = await reader.readline()
    print(f"收到: {data.decode()}")

    # 关闭连接
    writer.close()
    await writer.wait_closed()

asyncio.run(tcp_client())

asyncio TCP 服务器

import asyncio

async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"连接: {addr}")

    while True:
        data = await reader.readline()
        if not data:
            break
        writer.write(data)  # 回显
        await writer.drain()

    writer.close()
    await writer.wait_closed()

async def main():
    server = await asyncio.start_server(handle_client, 'localhost', 8080)
    print("服务器启动...")

    async with server:
        await server.serve_forever()

asyncio.run(main())

asyncio HTTP 请求

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:
        # 并发请求
        urls = ['http://httpbin.org/delay/1',
                'http://httpbin.org/delay/2',
                'http://httpbin.org/delay/3']
        results = await asyncio.gather(*[fetch(session, url) for url in urls])
        print(f"完成 {len(results)} 个请求")

asyncio.run(main())

asyncio UDP

import asyncio

async def udp_server():
    transport, protocol = await asyncio.create_server(
        lambda: UDPProtocol(),
        'localhost', 8081
    )
    async with transport:
        await transport.serve_forever()

class UDPProtocol(asyncio.Protocol):
    def datagram_received(self, data, addr):
        print(f"收到: {data.decode()}")
        self.transport.sendto(data, addr)

Task 任务管理

import asyncio

async def worker(name):
    await asyncio.sleep(1)
    return f"{name} 完成"

async def main():
    # 创建任务
    task1 = asyncio.create_task(worker('A'))
    task2 = asyncio.create_task(worker('B'))

    # 等待完成
    result1 = await task1
    result2 = await task2

    print(result1, result2)

asyncio.run(main())

超时处理

import asyncio

async def slow_task():
    await asyncio.sleep(10)
    return "完成"

async def main():
    try:
        result = await asyncio.wait_for(slow_task(), timeout=5)
    except asyncio.TimeoutError:
        print("任务超时")

asyncio.run(main())

并发控制

import asyncio

async def limited_concurrent(tasks, limit):
    semaphore = asyncio.Semaphore(limit)

    async def wrapped(task):
        async with semaphore:
            return await task

    return await asyncio.gather(*[wrapped(t) for t in tasks])

# 限制最大并发数
tasks = [fetch(url) for url in urls]
results = await limited_concurrent(tasks, 10)  # 最大10个并发

asyncio vs 多线程

异步 IO 模型

同步 IO: 发起请求 -> 等待 -> 处理响应（阻塞）
异步 IO: 发起请求 -> 继续执行 -> 回调处理（非阻塞）
协程: await 挂起 -> 其他协程执行 -> await 返回

协程关键字

要点总结

• async def 定义协程，await 挂起等待
• asyncio.run() 运行主协程
• asyncio.gather() 并发执行多个协程
• asyncio.open_connection() 创建 TCP 连接
• asyncio.start_server() 创建 TCP 服务器
• aiohttp 是异步 HTTP 客户端库
• asyncio.Semaphore 控制并发数量
• asyncio.wait_for() 设置超时
• asyncio 适合高并发 IO 密集型场景
• 协程单线程高效，不受 GIL 限制