Python asyncio 运行速度较慢

Question

我是 Python 和并行执行和异步的新手。我做错了吗？我的代码运行速度较慢（或最多等于）以传统方式运行脚本所需的时间，没有异步。

import asyncio, os, time, pandas as pd
start_time = time.time()

async def main():
    coroutines = list()
    for root, dirs, files in os.walk('.', topdown=True):
        for file in files:
            coroutines.append(cleaner(file))
        await asyncio.gather(*coroutines)

async def cleaner(file):
 df = pd.read_csv(file, sep='\n', header=None, engine='python', quoting=3)
 df = df[0].str.strip(' \t"').str.split('[,|;: \t]+', 1, expand=True).rename(columns={0: 'email', 1: 'data'}) 
 df[['email', 'data']].to_csv('x1', sep=':', index=False, header=False, mode='a', compression='gzip')


asyncio.run(main())
print("--- %s seconds ---" % (time.time() - start_time))

Answer 1

您的工作负载似乎是读取文件 --> 使用 pandas 处理 --> 写入文件。这是多处理的理想选择，因为每个工作项都非常独立。 pandas 读取/写入文件系统的例程，就像任何阻塞操作一样，不适合使用 asyncio，除非您在 asyncio 的线程或进程池中运行它们。

相反，这些多个操作是真正的并行执行的良好候选者，而 asyncio 没有提供给您。 （它的线程和进程池也是不错的选择）。

import multiprocessing as mp
import os

def walk_all_files(path):
    for root, dirs, files in os.walk('.', topdown=True):
        for file in files:
            yield os.path.join(root, file)

def cleaner(path):
    return "sparkly"

def clean_all(path="."):
    files = list(walk_all_files(path))
    # using cpu*2 assuming that there is a lot of cpu heavy
    # work that can be done by some processes while others
    # wait on I/O. This is only a guess.
    cpu_count = min(len(files), mp.cpu_count()*2)
    with mp.Pool(cpu_count) as pool:
        # assuming processing is fairly long but also kindof random depending on
        # file contents, setting chunksize to 1 so that subprocess gets new work
        # item from parent on each round. You could set it higher to have fewer
        # interactions between parent and worker.
        result = pool.map(cleaner, files, chunksize=1)

if __name__ == "__main__":
    clean_all(".")