为什么concurrent.futures在将结果附加到列表时返回空文件？

Question

我试图使用concurrent.futures读取几个.json文件，并将结果存储在列表中。下面是我的代码：

data = []
PATH_TO_JSON = '/media/My Passport/echo2_1'

def jsonread (file):
    path = PATH_TO_JSON + '/' + file
    with open(path, 'r', encoding='utf-8') as f:
        for line in f:
            data.append(json.loads(line.rstrip('\n|\r')))


with concurrent.futures.ProcessPoolExecutor() as executor:
    files = [file for file in listdir(PATH_TO_JSON)]
    results = executor.map(jsonread, files)

但是代码返回一个空的data列表。我发现了一个类似的问题。但我无法将答案应用到我的工作中。我很感谢你的详细回答，因为我是multiprocessing和concurrent.futures的新手

Answer 1

每个进程都将从它们自己的内存开始。这使得在全局列表中共享数据变得困难。相反，请尝试重构jsonread方法以返回列表

def jsonread(file):
    data = []
    path = PATH_TO_JSON + '/' + file
    with open(path, 'r', encoding='utf-8') as f:
        for line in f:
            data.append(json.loads(line.rstrip('\n|\r')))
    return data

对于多进程代码，必须确保Python解释器可以导入主模块而不会产生副作用。每个进程将启动一个解释器，该解释器将加载主模块，因此必须确保只有主进程（第一个解释器）启动ProcessPoolExecutor。重构代码，在with语句周围包含一个主保护：

if __name__ == '__main__':
    with concurrent.futures.ProcessPoolExecutor() as executor:
        files = [file for file in listdir(PATH_TO_JSON)]
        results = executor.map(jsonread, files)

如果我们要在最后打印results的值，我们得到：

print(results)
<generator object _chain_from_iterable_of_lists at 0x000001B45FD4A200>

你在这里看到的是我们有一个发电机。python中的生成器是懒惰的，它们在调用next之前不会做任何事情。因为我们更改了jsonread以返回字符串列表，所以我们还必须将它们连接在一起。在jsonread中的值上循环将隐式调用生成器上的next。最终代码：

PATH_TO_JSON = '/media/My Passport/echo2_1'


def jsonread(file):
    data = []
    path = PATH_TO_JSON + '/' + file
    with open(path, 'r', encoding='utf-8') as f:
        for line in f:
            data.append(json.loads(line.rstrip('\n|\r')))
    return data


if __name__ == '__main__':
    with concurrent.futures.ProcessPoolExecutor() as executor:
        files = [file for file in listdir(PATH_TO_JSON)]
        results = executor.map(jsonread, files)
        data = [line for result in results for line in result]