《用 asyncio 构建异步任务队列：Python 并发编程的实战与思考》

一、引言：并发编程的新时代

在现代软件开发中，性能已不再是锦上添花，而是产品成功的基石。尤其在 I/O 密集型场景中，如网络爬虫、实时数据处理、微服务通信等，传统的同步编程模式往往力不从心。

Python 虽然因其简洁优雅的语法和强大的生态系统广受欢迎，但其并发能力常常被误解。事实上，随着 asyncio 的引入，Python 在异步编程领域焕发出新的生命力。

本文将带你从零构建一个基于 asyncio 的异步任务队列，并结合实际案例展示其在高并发场景中的应用价值。无论你是刚入门的开发者，还是追求性能优化的架构师，都能在这篇文章中找到灵感与实用技巧。

---

二、asyncio 简介：Python 异步编程的核心

asyncio 是 Python 3.4 引入的标准库，用于编写异步 I/O 代码。它基于事件循环机制，允许我们以非阻塞方式执行任务，从而提升程序的并发能力。

核心概念：

• 事件循环（Event Loop）：调度和执行协程的核心机制。
• 协程（Coroutine）：使用 async def 定义的函数，可通过 await 暂停执行。
• 任务（Task）：事件循环中的执行单元，包装协程以便调度。
• Future：表示尚未完成的异步操作结果。

---

三、构建异步任务队列：从原理到实现

我们将逐步构建一个可复用的异步任务队列，具备任务调度、并发执行、异常处理和结果回调等功能。

1. 设计目标

• 支持异步任务的提交与调度
• 控制并发数量（限流）
• 支持任务结果回调
• 具备异常捕获机制

---

四、基础实现：异步任务队列的骨架

我们先构建一个最小可用的异步任务队列。

import asyncio
from typing import Callable, Anyclass AsyncTaskQueue:def __init__(self, max_concurrency: int = 5):self.semaphore = asyncio.Semaphore(max_concurrency)self.queue = asyncio.Queue()self.running = Falseasync def worker(self):while self.running:func, args, kwargs, callback = await self.queue.get()async with self.semaphore:try:result = await func(*args, **kwargs)if callback:await callback(result)except Exception as e:print(f"任务执行异常：{e}")finally:self.queue.task_done()<