如何用Python多进程并行执行PyType？及进程重复处理文件问题

我有数千个Python文件需要执行PyType类型检查，单线程执行预计耗时数月，因此尝试使用Python multiprocessing创建多进程，每个进程单独处理一个文件以提升效率。但编写的脚本运行后发现，所有进程均在处理同一个文件而非不同文件，我怀疑这与PyType的内部运行机制有关，请问是否有人遇到过类似问题？

实现脚本如下：

import multiprocessing
import subprocess
import os

def work(cmd):
    result = subprocess.run(cmd, capture_output=True, text=True)
    return f"{os.getpid()} - Command: {cmd}\nOutput: {result.stdout}\nError: {result.stderr}"


if __name__ == '__main__':
    my_dir = "xxx"
    cmds = []
    for file in os.listdir(my_dir):
        if file.endswith(".py"):
            cmds.append(["pytype", "-o", os.path.join(my_dir, '.pytype'), os.path.join(my_dir, file)])
    count = multiprocessing.cpu_count()
    with multiprocessing.Pool(processes=count) as pool:
        results = pool.map(work, cmds)
    for result in results:
        print(result)

问题原因与解决办法 #

这个问题和PyType内部机制无关，核心问题是所有进程共享同一个输出缓存目录：你给所有PyType命令指定了同一个-o .pytype目录，PyType在处理时会依赖这个目录的缓存文件，多进程同时读写该目录会引发冲突，导致实际执行逻辑异常，看起来像是重复处理同一个文件。

解决方式1：给每个进程分配独立的输出目录

为每个待检查的Python文件创建单独的输出目录，避免多进程读写冲突：

if __name__ == '__main__':
    my_dir = "xxx"
    cmds = []
    for file in os.listdir(my_dir):
        if file.endswith(".py"):
            # 基于文件名生成唯一输出目录
            output_dir = os.path.join(my_dir, f'.pytype_{file[:-3]}')
            # 确保目录存在，不存在则创建
            os.makedirs(output_dir, exist_ok=True)
            cmds.append(["pytype", "-o", output_dir, os.path.join(my_dir, file)])
    count = multiprocessing.cpu_count()
    with multiprocessing.Pool(processes=count) as pool:
        results = pool.map(work, cmds)
    for result in results:
        print(result)

解决方式2：禁用PyType缓存

使用--no-cache参数关闭缓存功能，这样PyType不会读写共享缓存目录，自然不会出现冲突：

cmds.append(["pytype", "--no-cache", os.path.join(my_dir, file)])

额外验证步骤

在生成cmds列表后，可以打印前几个命令确认：

print(cmds[:3])

确保每个命令对应的目标Python文件路径是不同的，排除列表生成时的逻辑错误。

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