这是我的第476篇原创文章。

一、概念定义

1.1 并发是"结构"（多个任务如何组织），异步是"协作方式"（任务之间如何交互）

1.2 四种组合

组合	说明	例子
同步 + 串行	一件做完再做下一件，干等	单线程顺序执行
同步 + 并发	多线程同时执行，但每个线程内是同步阻塞的	10个线程各发一个HTTP请求，每个线程内requests.get()阻塞等待
异步 + 串行	单线程内用异步，但任务依次执行	asyncio里用await逐个调用，不并发
异步 + 并发	单线程内同时发起多个异步操作，通过事件循环调度	asyncio.gather(task1, task2, task3)

最常用的高效模式是"异步 + 并发"：一个线程里同时挂起多个IO操作，事件循环哪个就绪处理哪个。

1.3 核心联系

• 异步可以实现并发

• 并发不一定异步

• 异步不一定并发

• 并发是"同时处理多个任务"的能力，异步是"发起操作后不傻等"的策略。
  异步让单线程也能实现高并发IO，是现代高性能编程（如Python asyncio、Node.js）的核心思想。

1.4 同步和异步

同步调用（Synchronous Call）本质：线程阻塞在I/O等待

调用方发起请求后，必须阻塞等待被调用方处理完成并返回结果，才能继续执行后续代码。整个调用链在同一个线程/进程上下文中完成。

异步调用（Asynchronous Call） 本质：基于事件循环和I/O多路复用

调用方发起请求后立即返回（任务id或者其他内容），不等待最终结果，通过回调（通知）、或轮询机制在将来某个时间点获取处理结果。调用方和被调用方解耦，通常在不同线程/进程中执行。

二、生活化比喻

同步场景：餐厅点餐

你走到柜台前对服务员说："我要一份牛肉面"，然后站在柜台前一直等，直到厨师做好、服务员把面端给你，你才离开柜台去吃饭。这期间你不能做别的事。

特点：简单直接，但浪费时间（阻塞）

异步场景：外卖下单

你在APP上点完外卖，订单立即提交成功（接口返回）。然后你可以去工作、看电视，做其他事情。外卖员送餐后打电话通知你（回调），或者你时不时查看APP配送进度（轮询）。

特点：高效灵活，但流程更复杂

三、测试场景

测试场景：将100条工作记录写入数据库，需调用100次接口，每次耗时100ms

调用方式	线程数	总耗时	CPU利用率	内存消耗
同步串行	1	10,000ms	低（大量等待）	低
同步并行	10	1,000ms	高	中
异步回调	1	~100ms	极高	较高（事件对象）
消息队列	1+worker	~150ms	极高	高（队列缓冲）

结论：异步调用吞吐量是同步的10-100倍，但需要额外资源管理。

四、误区

❌ 误区1：异步一定比同步快
✅ 事实：单次调用异步反而更慢（有 overhead），优势在高并发

❌ 误区2：用async关键字就是异步
✅ 事实：必须配合事件循环和I/O等待，否则只是"伪异步"

❌ 误区3：异步不需要处理异常
✅ 事实：异常更难追踪，需统一异常处理和日志

❌ 误区4：异步调用可无限并发
✅ 事实：受连接池、文件描述符、内存限制

五、实现过程

方式1：基于线程实现回调异步

回调异步是指：启动一个耗时操作时，不等待结果，而是传入一个"回调函数"（类似外卖员），操作完成后自动调用该函数处理结果（打电话通知你）。回调函数的作用处理异步任务的结果，比如打印结果，将结果发邮件通知给某个人等等。

单线程异步IO：实现方式就是在主线程后面起一个线程（单个worker），执行耗时的任务（异步任务，如财务报告生成任务，单次调用），完成之后调用回调函数将结果进行通知。

import time
import threading
from datetime import datetime

def run(name, duration, callback):
    print(f"{datetime.now().strftime('%H:%M:%S')} - {name}: 开始执行")
    time.sleep(duration)  # 模拟耗时操作
    result = f"{name} 完成，耗时 {duration} 秒"
    callback(result)  # 执行回调


def async_task(name, duration, callback):
    """模拟异步任务：在后台线程执行，完成后调用回调"""

    response = "请稍后查看结果"
    # 在新线程中运行，不阻塞主程序
    thread = threading.Thread(target=run, args=(name, duration, callback))
    thread.start()

    return response

def callback_handler(result):
    """回调函数：处理异步任务的结果"""
    print(f"{datetime.now().strftime('%H:%M:%S')} - 收到结果: {result}")

# 主程序
print(f"{datetime.now().strftime('%H:%M:%S')} - 主程序开始")
print("启动异步任务...")

# 启动异步任务，立刻返回一个结果（不阻碍主线程），耗时的操作在后台（线程）执行，执行完成会调用回调函数打印结果
result = async_task("任务A", 20, callback_handler)
print(result)

print(f"{datetime.now().strftime('%H:%M:%S')} - 主程序可以做其他事情")
time.sleep(4)  # 等待所有任务完成
print(f"{datetime.now().strftime('%H:%M:%S')} - 主程序结束")

多线程（并发）异步IO：对于多个耗时的任务（如报告生成、报告解析等），每一个任务都在后台起一个线程（多个worker），完成之后调用回调函数。

import time
import threading
from datetime import datetime

def async_task(name, duration, callback):
    """模拟异步任务：在后台线程执行，完成后调用回调"""
    def run():
        print(f"{datetime.now().strftime('%H:%M:%S')} - {name}: 开始执行")
        time.sleep(duration)  # 模拟耗时操作
        result = f"{name} 完成，耗时 {duration} 秒"
        callback(result)  # 执行回调

    # 在新线程中运行，不阻塞主程序
    thread = threading.Thread(target=run)
    thread.start()

def callback_handler(result):
    """回调函数：处理异步任务的结果"""
    print(f"{datetime.now().strftime('%H:%M:%S')} - 收到结果: {result}")

# 主程序
print(f"{datetime.now().strftime('%H:%M:%S')} - 主程序开始")
print("启动多个异步任务...")

# 启动3个异步任务，不会阻塞
async_task("任务A", 2, callback_handler)  # 耗时2秒
async_task("任务B", 1, callback_handler)  # 耗时1秒
async_task("任务C", 3, callback_handler)  # 耗时3秒

print(f"{datetime.now().strftime('%H:%M:%S')} - 所有任务已启动，主程序继续执行")
time.sleep(0.5)
print(f"{datetime.now().strftime('%H:%M:%S')} - 主程序可以做其他事情")
time.sleep(4)  # 等待所有任务完成
print(f"{datetime.now().strftime('%H:%M:%S')} - 主程序结束")

方式2：基于协程（async/await）实现

现代 async/await 语法糖，底层就是 Future，但更简洁

1. async def 定义协程函数

2. await  挂起协程等待结果

3. asyncio.create_task() 创建后台任务

4. asyncio.gather() 并发执行多个任务

5. asyncio.run()  启动事件循环,专门用来启动和运行异步代码

单协程异步IO：实现方式就是在后台起一个协程，执行耗时的任务（如财务报告生成任务），完成之后调用回调函数将结果进行通知。

import asyncio
from datetime import datetime
async def modern_async():
    # async 函数自动返回 Future（Task）
    task1 = asyncio.create_task(asyncio.sleep(10, result="耗时任务完成"))
    print(f"[{datetime.now()}] 任务已创建，请稍后查看结果")
    # await 等待 Future 完成
    result1 = await task1
    print(f"[{datetime.now()}] {result1}")
if __name__ == '__main__':
    asyncio.run(modern_async())

结果：

异步 + 串行

import asyncio
from datetime import datetime
async def modern_async():
    # async 函数自动返回 Future（Task）
    task1 = asyncio.create_task(asyncio.sleep(5, result="耗时任务1完成"))
    task2 = asyncio.create_task(asyncio.sleep(10, result="耗时任务2完成"))
    print(f"[{datetime.now()}] 任务已创建，继续执行...")
    # await 等待 Future 完成
    result1 = await task1
    print(f"[{datetime.now()}] {result1}")
    result2 = await task2
    print(f"[{datetime.now()}] {result2}")
if __name__ == '__main__':
    asyncio.run(modern_async())

结果：

await 只挂起当前协程，事件循环继续调度其他 Task（task2 一直在跑）。两个 Task 早已同时创建并启动，只是获取结果的顺序是先1后2。即使交换 await 顺序，总耗时仍是10秒。

异步 + 并发

import asyncio
from datetime import datetime
async def modern_async():
    # async 函数自动返回 Future（Task）
    task1 = asyncio.create_task(asyncio.sleep(5, result="耗时任务1完成"))
    task2 = asyncio.create_task(asyncio.sleep(10, result="耗时任务2完成"))
    tasks = [task1, task2]
    print(f"[{datetime.now()}] 任务已创建，继续执行...")
    # await 等待 Future 完成
    results = await asyncio.gather(*tasks)
    print(f"[{datetime.now()}] {results}")

if __name__ == '__main__':
    asyncio.run(modern_async())

结果：

方式3：基于异步架构（消息队列）实现

使用 Redis + RQ (Redis Queue)

安装依赖：

pip install redis rq
# 需要本地运行 Redis 服务器
# macOS: brew install redis && brew services start redis
# Ubuntu: sudo apt install redis-server && sudo systemctl start redis

创建异步任务 (tasks.py)：

import time
from datetime import datetime

def send_email(to, subject, body):
    """
    模拟耗时的邮件发送任务
    """
    print(f"[{datetime.now()}] 开始发送邮件给 {to}")
    time.sleep(5)  # 模拟耗时操作
    print(f"[{datetime.now()}] 邮件发送完成: {subject}")
    return f"邮件已发送至 {to}"

def process_image(image_path):
    """
    模拟图片处理任务
    """
    print(f"[{datetime.now()}] 开始处理图片: {image_path}")
    time.sleep(3)
    print(f"[{datetime.now()}] 图片处理完成")
    return f"图片 {image_path} 已处理"

def generate_report(user_id):
    """
    模拟报告生成任务
    """
    print(f"[{datetime.now()}] 开始生成用户 {user_id} 的报告")
    time.sleep(10)
    print(f"[{datetime.now()}] 报告生成完成")
    return f"用户 {user_id} 的报告已生成"

创建任务生产者 (producer.py)：

from redis import Redis
from rq import Queue
from tasks import send_email, process_image, generate_report
import time
from datetime import datetime

# 连接 Redis
redis_conn = Redis(host='localhost', port=6379, db=0)
q = Queue(connection=redis_conn)

def main():
    print(f"[{datetime.now()}] 主程序开始")

    # 异步发送邮件 - 立即返回，不等待完成
    job1 = q.enqueue(send_email, 'user@example.com', '欢迎注册', '感谢您的注册！')
    print(f"[{datetime.now()}] 邮件任务已提交 (job id: {job1.id})")

    # 异步处理图片
    job2 = q.enqueue(process_image, '/path/to/image.jpg')
    print(f"[{datetime.now()}] 图片处理任务已提交 (job id: {job2.id})")

    # 异步生成报告
    job3 = q.enqueue(generate_report, user_id=12345)
    print(f"[{datetime.now()}] 报告生成任务已提交 (job id: {job3.id})")

    # 主程序可以继续执行其他逻辑，不用等待任务完成
    print(f"[{datetime.now()}] 主程序继续执行其他工作...")
    time.sleep(2)
    print(f"[{datetime.now()}] 主程序结束")

    # 可选：等待并获取结果（通常不推荐，破坏异步性）
    # print(f"任务结果: {job1.result}")

if __name__ == '__main__':
    main()

终端1 - 运行生产者:

python producer.py

输出：

[2026-01-15 10:30:15] 主程序开始
[2026-01-15 10:30:15] 邮件任务已提交 (job id: 6c3b2a1c-...)
[2026-01-15 10:30:15] 图片处理任务已提交 (job id: 7d4c3b2d-...)
[2026-01-15 10:30:15] 报告生成任务已提交 (job id: 8e5d4c3e-...)
[2026-01-15 10:30:15] 主程序继续执行其他工作...
[2026-01-15 10:30:17] 主程序结束

终端2 - 运行消费者

# 监听 default 队列
rq worker --with-scheduler

# 或者监听指定队列
rq worker high default low

输出：

[2026-01-15 10:30:16] 开始发送邮件给 user@example.com
[2026-01-15 10:30:21] 邮件发送完成: 欢迎注册
[2026-01-15 10:30:21] 开始处理图片: /path/to/image.jpg
[2026-01-15 10:30:24] 图片处理完成
[2026-01-15 10:30:24] 开始生成用户 12345 的报告
[2026-01-15 10:30:34] 报告生成完成

补充：

QPS=并发数/响应时间

如果我有1000个质检任务，每个质检任务（一个任务调一次api）耗时10s，我想10s左右处理完，那我就得开1000个线程（并发），QPS达到1000/10=100,这显然资源扛不住，还有就是有段时间突然1000个请求过来，有段时间没用请求过来，如何控制QPS比较平稳（削峰）？

可以通过异步实现削峰动态调整并发度

六、小结

异步任务不一定是并发的，它们的关系取决于具体实现。异步 vs 并发的核心区别

概念	含义	关键特征
异步 (Async)	提交任务后立即返回，不等待结果	非阻塞、回调、事件循环
并发 (Concurrent)	多个任务在同一时间段内执行	可以是并行，也可以是交替执行

模式1：异步但不并发（单线程异步IO、单worker消息队列）

模式2：异步且并发（多进程/多线程异步IO、多worker消息队列）

消息队列提供了异步的架构，而并发能力需要通过水平扩展 Worker 进程来实现

# 消息队列提供：异步 + 解耦 + 可靠传输
q.enqueue(task1)# 立即返回
q.enqueue(task2)# 立即返回

# 并发需要：手动扩展 worker 数量
# Worker数量 = 最大并发数

作者简介：

读研期间发表6篇SCI数据挖掘相关论文，现在某研究院从事数据算法相关科研工作，结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创，以最简单的方式理解和学习，关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。