使用asyncio计算文本距离度量时异步与同步耗时一致的问题排查

问题根源 #

你的代码里的异步函数都是假异步：所有metric_xx函数只是包装了textdistance的同步计算逻辑，内部没有任何await异步IO操作。asyncio的事件循环是单线程的，只有当任务遇到IO阻塞（比如网络请求、文件读写）时才会切换到其他任务。而文本距离计算是CPU密集型操作，会一直占用线程，事件循环根本没机会切换任务，所以40个metric任务其实是串行执行的，和同步模式耗时完全一致。

用asyncio实现的正确方式 #

要让asyncio发挥作用，必须把CPU密集型的计算放到线程池/进程池里执行，让asyncio的事件循环可以在等待计算结果时处理其他任务。Python 3.9+提供了asyncio.to_thread()，可以方便地把同步函数放到线程池运行；也可以用loop.run_in_executor()自定义线程/进程池。

修改后的代码示例 #

第一步：简化metric定义，直接用同步函数

不需要写40个重复的async函数，直接把textdistance的方法存成列表：

import textdistance as td
import asyncio
import nest_asyncio
nest_asyncio.apply()

s1 = ["string1", "string2", ...]  # 50万+字符串
s2 = ["string1", "string2", ...]

# 直接定义所有需要调用的同步方法
metrics = [
    td.hamming.normalized_distance,
    td.mlipns.normalized_distance,
    td.levenshtein.normalized_distance,
    td.damerau_levenshtein.normalized_distance,
    td.jaro_winkler.normalized_distance,
    td.strcmp95.normalized_distance,
    td.needleman_wunsch.normalized_distance,
    td.gotoh.normalized_distance,
    td.smith_waterman.normalized_distance,
    td.jaccard.normalized_distance,
    td.sorensen_dice.normalized_distance,
    td.tversky.normalized_distance,
    td.overlap.normalized_distance,
    td.tanimoto.normalized_distance,
    td.cosine.normalized_distance,
    td.monge_elkan.normalized_distance,
    td.bag.normalized_distance,
    td.lcsseq.normalized_distance,
    td.lcsstr.normalized_distance,
    td.ratcliff_obershelp.normalized_distance,
    td.hamming.normalized_similarity,
    td.mlipns.normalized_similarity,
    td.levenshtein.normalized_similarity,
    td.damerau_levenshtein.normalized_similarity,
    td.jaro_winkler.normalized_similarity,
    td.strcmp95.normalized_similarity,
    td.needleman_wunsch.normalized_similarity,
    td.gotoh.normalized_similarity,
    td.smith_waterman.normalized_similarity,
    td.jaccard.normalized_similarity,
    td.sorensen_dice.normalized_similarity,
    td.tversky.normalized_similarity,
    td.overlap.normalized_similarity,
    td.tanimoto.normalized_similarity,
    td.cosine.normalized_similarity,
    td.monge_elkan.normalized_similarity,
    td.bag.normalized_similarity,
    td.lcsseq.normalized_similarity,
    td.lcsstr.normalized_similarity,
    td.ratcliff_obershelp.normalized_similarity,
]

第二步：用asyncio.to_thread包装同步计算

在计算时，把每个metric的调用放到线程池里，让事件循环可以并行处理多个任务：

async def calculate_single_pair(sent1, sent2):
    tasks = []
    for metric in metrics:
        # 用to_thread把同步计算放到线程池执行
        task = asyncio.to_thread(metric, sent1, sent2)
        tasks.append(task)
    return await asyncio.gather(*tasks)

async def calc(s1, s2):
    # 同时处理多对字符串，进一步提升并行度
    pair_tasks = [calculate_single_pair(s1[i], s2[i]) for i in range(len(s1))]
    return await asyncio.gather(*pair_tasks)

result = asyncio.run(calc(s1, s2))

进阶：自定义进程池处理CPU密集型任务

因为文本计算是CPU密集型，线程池受GIL限制，并行效率有限。可以用concurrent.futures.ProcessPoolExecutor来创建进程池，绕过GIL：

from concurrent.futures import ProcessPoolExecutor

async def calc(s1, s2):
    # 创建进程池，进程数建议设为CPU核心数
    with ProcessPoolExecutor(max_workers=4) as executor:
        loop = asyncio.get_running_loop()
        pair_tasks = []
        for i in range(len(s1)):
            # 每个字符串对的所有metric任务放到进程池
            metric_tasks = [loop.run_in_executor(executor, metric, s1[i], s2[i]) for metric in metrics]
            pair_result = asyncio.gather(*metric_tasks)
            pair_tasks.append(pair_result)
        return await asyncio.gather(*pair_tasks)

result = asyncio.run(calc(s1, s2))

关键说明 #

• asyncio本身不适合纯CPU密集型任务，但结合线程池/进程池，可以让asyncio负责任务调度，把计算任务放到池子里并行执行，从而提升效率。
• 进程池适合CPU密集型任务（绕过GIL），线程池适合IO密集型+少量CPU的任务。对于文本距离计算，进程池的效率会比线程池更高。
• 注意内存占用：50万条数据+40个metric，结果会非常大，建议考虑分批处理或者流式输出，避免内存溢出。

内容的提问来源于stack exchange，提问作者Atilio