为何这段asyncio代码无法实现并发执行？

你的代码串行执行的核心原因是：convert_from_path是同步阻塞IO操作，直接在async协程中调用会阻塞整个asyncio事件循环。asyncio的事件循环基于单线程调度，一旦某个协程执行阻塞代码，其他协程会被暂停，直到阻塞操作完成，最终表现为串行执行。

步骤1：修复代码语法错误 #

先修正代码里的两处语法问题：

• 去掉@click.option行末尾的冒号（语法错误）
• 将for path in path改为for path in paths（遍历对象错误）

步骤2：将同步阻塞函数异步化 #

通过把convert_from_path放到线程池（或进程池）中执行，让asyncio事件循环可以同时调度其他协程，实现真正的并发。以下是两种常用方案：

方案一：使用asyncio.to_thread（Python 3.9+） #

asyncio.to_thread是Python 3.9新增的API，专门用于将同步函数异步执行，内部封装了线程池：

import click
import asyncio
from pdf2image import convert_from_path
from functools import wraps

def coro(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        return asyncio.run(f(*args, **kwargs))

    return wrapper

async def my_coroutine(path):
    print(path)
    # 将同步阻塞函数放到线程池执行，释放事件循环
    return await asyncio.to_thread(
        convert_from_path,
        path,
        fmt="ppm",
        poppler_path=""  # 替换为你的poppler实际路径，如Windows下的r"C:\poppler-24.02.0\bin"
    )

@click.command()
@click.option("-s", "settings_path", required=False, type=str)
@coro
async def dlr(settings_path) -> None:
    paths = [...]  # 替换为你的PDF路径列表

    # 并发执行所有协程
    responses = await asyncio.gather(*[my_coroutine(path) for path in paths])
    # 后续可处理转换后的图片，比如保存到本地等

@click.group()
def cli() -> None:
    pass

cli.add_command(dlr)

if __name__ == "__main__":
    cli()

方案二：使用concurrent.futures（兼容Python 3.9以下） #

如果你的Python版本低于3.9，可以用ThreadPoolExecutor配合事件循环的run_in_executor方法：

import click
import asyncio
from pdf2image import convert_from_path
from functools import wraps
from concurrent.futures import ThreadPoolExecutor

def coro(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        return asyncio.run(f(*args, **kwargs))

    return wrapper

# 全局线程池，可自定义大小，比如max_workers=4
executor = ThreadPoolExecutor(max_workers=4)

async def my_coroutine(path):
    print(path)
    loop = asyncio.get_running_loop()
    # 提交同步任务到线程池执行
    return await loop.run_in_executor(
        executor,
        convert_from_path,
        path,
        fmt="ppm",
        poppler_path=""
    )

@click.command()
@click.option("-s", "settings_path", required=False, type=str)
@coro
async def dlr(settings_path) -> None:
    paths = [...]

    responses = await asyncio.gather(*[my_coroutine(path) for path in paths])

@click.group()
def cli() -> None:
    pass

cli.add_command(dlr)

if __name__ == "__main__":
    cli()

额外说明 #

如果PDF转换属于CPU密集型操作（比如大文件批量转换），可以将ThreadPoolExecutor替换为ProcessPoolExecutor，利用多进程实现并行，但线程池对IO密集型场景（如文件读取、外部工具调用）的开销更低，更适合当前场景。

内容的提问来源于stack exchange，提问作者Tlaloc-ES