生成假用户档案：Python异步代码为何比同步代码更慢？

你测试了同步和异步两种生成10000条Faker用户档案的代码，结果显示异步耗时（13-19秒）远高于同步（7.8-8.6秒），以下是核心原因及针对Django APIView的解决方案：

你的测试代码 #

异步实现 #

from faker import Faker
import time
import asyncio
from pprint import pprint

generator = Faker()

async def main():
    tasks = [loop.run_in_executor(None, generator.profile) for _ in range(10000)]
    await asyncio.gather(*tasks)
    return [t.result() for t in tasks]

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    times = []
    for i in range(10):
        tic = time.time()
        r = loop.run_until_complete(main())
        times.append(time.time() - tic)
    pprint(times)

同步实现 #

from timeit import timeit, repeat
import time
from pprint import pprint

setup = '''
from faker import Faker
generator = Faker()
'''
statement = '''
[generator.profile() for _ in range(10000)]
'''
pprint(repeat(setup = setup, stmt = statement, number = 1, repeat = 10))

异步代码更慢的核心原因 #

• Faker是CPU密集型任务，异步无优势：generator.profile()是纯CPU计算的同步函数，没有IO等待（比如网络请求、数据库查询）。异步编程的优势在于IO等待时切换任务，但CPU密集型任务会持续占用线程，反而因为线程调度、上下文切换产生额外开销。
• 线程池调度成本过高：你的异步代码给每个profile()都创建了一个线程任务，默认线程池大小等于CPU核心数（比如4核仅能同时执行4个任务）。10000个任务需要频繁调度切换，相比同步单线程连续执行，额外开销被放大。
• 计时方式的差异：同步代码用timeit.repeat，它会自动优化初始化开销，计时更精准；异步代码用time.time()手动计时，包含了事件循环初始化、线程池创建的额外时间，进一步拉大差距。

针对Django APIView的解决方案 #

方案1：保持同步实现（推荐） #

由于Faker是CPU密集型任务，同步实现本身效率更高，直接在APIView中使用即可：

from rest_framework.views import APIView
from rest_framework.response import Response
from faker import Faker

generator = Faker()

class GenerateUserDataView(APIView):
    def get(self, request):
        count = int(request.query_params.get('count', 100))
        profiles = [generator.profile() for _ in range(count)]
        return Response(profiles)

方案2：大数量请求用Celery后台处理 #

如果用户需要生成十万级以上数据，直接在APIView处理会导致请求超时，此时用Celery把任务放到后台执行：

# tasks.py
from celery import shared_task
from faker import Faker

generator = Faker()

@shared_task
def generate_user_profiles(count):
    return [generator.profile() for _ in range(count)]

# views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from .tasks import generate_user_profiles

class GenerateUserDataView(APIView):
    def get(self, request):
        count = int(request.query_params.get('count', 100))
        task = generate_user_profiles.delay(count)
        return Response({"task_id": task.id, "status": "pending"})

方案3：优化异步用法（仅作演示，不推荐） #

如果一定要用异步，可通过批量任务减少调度开销，但本质不会比同步更快：

import asyncio
from faker import Faker

generator = Faker()

async def generate_batch(count):
    return [generator.profile() for _ in range(count)]

async def main():
    batch_size = 10000 // 4  # 按CPU核心数拆分
    tasks = [asyncio.to_thread(generate_batch, batch_size) for _ in range(4)]
    results = await asyncio.gather(*tasks)
    return [item for batch in results for item in batch]

内容的提问来源于stack exchange，提问作者Amundeep Singh Dhaliwal