python - 使用 ProcessPoolExecutor 时更新变量

Question

if __name__ == '__main__':

    MATCH_ID = str(doc_ref2.id)

    MATCH_ID_TEAM = doc_ref3.id

    with concurrent.futures.ProcessPoolExecutor(max_workers=30) as executor:
        results = list(executor.map(ESPNlayerFree, teamList1))
    
    MATCH_ID_TEAM = str(doc_ref4.id)

    with concurrent.futures.ProcessPoolExecutor(max_workers=30) as executor:
        results = list(executor.map(ESPNlayerFree, teamList2))

当我打印 MATCH_ID_TEAM 时，它会打印值。但在此过程中，它显示了我在顶部设置为空的默认值。

如何将我的变量值更新到所有进程？

ESPNPlayerFree is a class that takes `id` as an argument. So `teamList1` and `teamList2` are list of ids to initialize my objects.

MATCH_ID 和 MATCH_ID_TEAM 是我的类 ESPNPlayerFree 中使用的变量

操作系统 Windows 10 64位

集成开发环境 Pycharm

Python版本 3.6.1

Answer 1

我正在寻找@furas 几天前在他的评论中留下的地方。最简单的方法确实是将您在类中需要的所有内容与 .map() 一起传递。 executor.map()期待可迭代对象，它会被压缩到一个参数元组中，用于在您的工作人员中进行的每个函数调用。

您显然需要 MATCH_ID 和 MATCH_ID_TEAM 在整个作业中保持相同，即对 executor 的一次 调用。 map ()。您面临的挑战是两者都是可迭代对象(字符串)，但您需要将它们作为一个整体进行复制，并且通常足以与您的 teamlist-iterable 中的每一项相匹配。

所以你要做的就是用 itertools.repeat() 包裹这些字符串当您将它们与团队 ID 列表一起传递给 .map() 时。 itertools.repeat() 默认返回传递对象的无限迭代器。 ProcessPoolExecutor 然后在内部使用 zip() 将来自所有可迭代对象的项目组合为参数。

import concurrent.futures
import multiprocessing
from itertools import repeat


class ESPNPlayerFree:
    def __init__(self, team_id, match_id, match_id_team):
        self.teams_id = team_id
        self.match_id = match_id
        self.match_id_team = match_id_team
        print(
            multiprocessing.current_process().name,
            self.teams_id, self.match_id, self.match_id_team
        )


if __name__ == '__main__':

    teams1 = [f"id{i}" for i in range (10)]
    teams2 = [f"id{i}" for i in range(10, 20)]

    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:

        MATCH_ID = 'doc_ref2.id'
        MATCH_ID_TEAM = 'doc_ref3.id'

        results = list(
            executor.map(
                ESPNPlayerFree,
                teams1,
                repeat(MATCH_ID),
                repeat(MATCH_ID_TEAM),
            )
        )

        print("--- new MATCH_ID_TEAM ---")
        MATCH_ID_TEAM = 'doc_ref4.id'

        results = list(
            executor.map(
                ESPNPlayerFree,
                teams2,
                repeat(MATCH_ID),
                repeat(MATCH_ID_TEAM),
            )
        )

输出:

ForkProcess-1 id0 doc_ref2.id doc_ref3.id
ForkProcess-2 id1 doc_ref2.id doc_ref3.id
ForkProcess-3 id2 doc_ref2.id doc_ref3.id
ForkProcess-4 id3 doc_ref2.id doc_ref3.id
ForkProcess-1 id4 doc_ref2.id doc_ref3.id
ForkProcess-3 id5 doc_ref2.id doc_ref3.id
ForkProcess-2 id6 doc_ref2.id doc_ref3.id
ForkProcess-4 id7 doc_ref2.id doc_ref3.id
ForkProcess-3 id8 doc_ref2.id doc_ref3.id
ForkProcess-1 id9 doc_ref2.id doc_ref3.id
--- new MATCH_ID_TEAM ---
ForkProcess-1 id10 doc_ref2.id doc_ref4.id
ForkProcess-3 id11 doc_ref2.id doc_ref4.id
ForkProcess-2 id12 doc_ref2.id doc_ref4.id
ForkProcess-4 id13 doc_ref2.id doc_ref4.id
ForkProcess-1 id14 doc_ref2.id doc_ref4.id
ForkProcess-3 id15 doc_ref2.id doc_ref4.id
ForkProcess-2 id16 doc_ref2.id doc_ref4.id
ForkProcess-4 id17 doc_ref2.id doc_ref4.id
ForkProcess-2 id18 doc_ref2.id doc_ref4.id
ForkProcess-1 id19 doc_ref2.id doc_ref4.id

Process finished with exit code 0

对于第二个工作，有了新的 MATCH_ID_TEAM，您就不必再次重新创建 ProcessPoolExecutor，您只需留在上下文管理器中再次使用现有的只要你需要它。