多处理池输出 <multiprocessing.pool.IMapIterator object at "Insert random letters/numbers">

Question

我正在尝试优化一个简单的抛硬币程序，该程序通过模拟抛硬币来计算正面和反面的概率（主要是为了学习多处理），并且我不断得到输出：。我知道我需要将每个对象设为整数，但我不知道该怎么做。

import random
from multiprocessing import Pool

def rolldie(): # 1 = heads, 2 = tails
    dices = int(random.randrange(1,3))
    return(int(dices))

def main():
    out = []
    if __name__ == '__main__':
        pool = Pool()
        for i in range(100): #this can be any number
            out.append(pool.imap(rolldie(), 0))
        return(out)

print(main())

Answer 1

您的代码有几个问题，我想与您一起审查。你可以使用方法multiprocessing.Pool.imap，但不能以你使用它的方式。

首先，需要移动您拥有的if __name__ == '__main__': 支票，以便对main() 的调用位于块内。否则，当您创建多处理池时，在那些使用spawn 创建新进程的平台上，每个新进程启动将在全局级别执行所有语句，包括print(main() 语句。但是，在您当前拥有 if __name__ == '__main__': 的位置后，main 将返回 N 次 None，其中 N 是池中正在启动的进程数。

接下来，你有pool.imap(rolldie(), 0)。 imap 的第一个参数应该指定一个函数。但在这里，您实际上是在调用该函数，因此作为第一个参数传递的是调用rolldie() 的返回值。这显然是不正确的。 imap 的第二个参数应该是一个iterable。因此，您的 rolldie 函数将为可迭代的每个元素调用一次，并将该元素作为参数传递。因此，rolldie 必须修改为采用单个参数。我们可以称它为 trial_number，rolldie 可以忽略它。 imap 的返回值本身就是一个iterable，它在迭代时会给出rolldie 返回的所有返回值。您可以迭代这个 iterable 并在返回结果时获取结果，或者一次将结果全部转换为列表：

import random
from multiprocessing import Pool

def rolldie(trial_number): # 1 = heads, 2 = tails
    dices = int(random.randrange(1,3))
    return(int(dices))

def main():
    pool = Pool()
    return list(pool.imap(rolldie, range(100)))

if __name__ == '__main__':
    print(main())

打印：

[2, 1, 2, 2, 2, 1, 2, 2, 1, 1, 2, 1, 1, 2, 1, 2, 2, 2, 2, 1, 1, 2, 1, 1, 1, 2, 2, 2, 1, 1, 1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 1, 2, 1, 2, 1, 1, 2, 2, 2, 1, 2, 2, 1, 1, 1, 1, 2, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 1, 2, 2, 2, 1, 2, 1, 2, 2, 1, 1, 1, 1, 1, 2, 2, 2, 2, 1, 2, 1, 2, 2, 2]

根据random.seed(a=None, version=2)上的文档：

如果a被省略或None，则使用当前系统时间。如果操作系统提供随机源，则使用它们而不是系统时间（有关可用性的详细信息，请参见 os.urandom() 函数）。

如果我再次运行上述代码，我可能会得到相同的确切结果，表明正在使用当前系统时间，并且池中的每个进程都在相同地初始化其随机数生成器，因此生成相同的随机数序列。 这不好！

相反，我们应该确保每个进程唯一地初始化其随机数生成器。 multiprocessing.Pool 构造函数的 initializer 参数指定了一个函数，该函数将为池中的每个进程调用一次以对其进行初始化。在这里，我们使用当前进程的进程 ID 为随机数生成器播种。

import random
from multiprocessing import Pool, current_process

def init_pool():
    random.seed(current_process().pid)

def rolldie(trial_number): # 1 = heads, 2 = tails
    dices = int(random.randrange(1,3))
    return(int(dices))

def main():
    pool = Pool(initializer=init_pool)
    return list(pool.imap(rolldie, range(100)))

if __name__ == '__main__':
    print(main())

打印：

[2, 2, 1, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 2, 1, 1, 2, 2, 1, 2, 2, 1, 2, 2, 2, 2, 1, 2, 1, 2, 2, 2, 1, 1, 1, 1, 1, 2, 1, 2, 1, 2, 1, 2, 2, 1, 2, 1, 1, 2, 2, 2, 2, 2, 2, 1, 1, 2, 2, 2, 2, 1, 1, 1, 2, 2, 1, 2, 2, 2, 2, 2, 1, 2, 2, 1, 2, 2, 2, 1, 2, 2, 1, 2, 2, 1, 1, 2, 1, 1, 2, 2, 2, 2, 2, 1, 1, 2, 1, 2]

Answer 2

您的工作函数不接受任何参数，因此您可以使用apply_async 来安排任务，然后使用get 结果：

（我添加了一些睡眠时间来模拟长时间运行的作业）

test.py：

import random
import time

from multiprocessing import Pool


def rolldie(idx):  # 1 = heads, 2 = tails
    t = random.uniform(1, 3)

    print(f"START: {idx} ({t:.3f}s)")

    time.sleep(t)
    i = random.randint(1, 2)

    print(f"END: {idx}")

    return i


def main():
    results, out = [], []

    with Pool() as pool:
        for i in range(10):
            results.append(pool.apply_async(rolldie, args=(i,)))

        for result in results:
            out.append(result.get())

    print(out)


if __name__ == "__main__":
    main()

测试：

$ python test.py
START: 0 (2.737s)
START: 1 (2.692s)
START: 2 (1.405s)
START: 3 (1.397s)
END: 3
START: 4 (2.537s)
END: 2
START: 5 (2.472s)
END: 1
START: 6 (2.373s)
END: 0
START: 7 (1.262s)
END: 5
START: 8 (1.416s)
END: 4
START: 9 (2.151s)
END: 7
END: 6
END: 8
END: 9
[2, 2, 2, 2, 2, 1, 1, 2, 2, 1]

我在运行测试的机器上只有 4 个 CPU，所以一开始只安排了 4 个作业，其余作业在队列中等待才能运行。