使用 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 位

IDE Pycharm

Python 版本 3.6.1

Answer 1

我想知道几天前@furas 在他的评论中留下的地方。最简单的方法确实是通过.map() 来传递你在课堂上需要的所有东西。 executor.map() 期待迭代，它被压缩到一个参数元组中，以便在您的工作人员中进行每个函数调用。

您显然需要MATCH_ID 和MATCH_ID_TEAM 才能在整个工作中保持不变，即一次 调用executor.map()。您面临的挑战是，这两个都是可迭代对象（字符串），但您需要将它们作为一个整体进行复制，并且通常足以与您的团队列表可迭代项中的每个项目相匹配。

因此，当您将这些字符串与团队 ID 列表一起传递给 .map() 时，您只需用 itertools.repeat() 包装这些字符串。 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，只要您需要，您只需在上下文管理器中再次使用现有的。