Python多处理：如何创建x个进程并返回返回值

Question

我有一个使用线程创建的程序，但后来我了解到线程不会在 python 中并发运行，而进程可以。结果，我试图使用多处理重写程序，但我很难这样做。我尝试了几个示例来展示如何创建进程和池，但我不认为这正是我想要的。

以下是我尝试过的代码。该程序试图通过在包含圆的图形上随机放置点来估计 pi ​​的值。该程序有两个命令行参数：一个是我要创建的线程/进程的数量，另一个是尝试在图表上放置的点的总数 (N)。

import math
import sys
from time import time
import concurrent.futures
import random
import multiprocessing as mp


def myThread(arg):
    # Take care of imput argument
    n = int(arg)
    print("Thread received. n = ", n)

    # main calculation loop
    count = 0
    for i in range (0, n):
        x = random.uniform(0,1)
        y = random.uniform(0,1)
        d = math.sqrt(x * x + y * y)
        if (d < 1):
            count = count + 1
    print("Thread found ", count, " points inside circle.")
    return count;
        
# end myThread

# receive command line arguments
if (len(sys.argv) == 3):
    N = sys.argv[1]  # original ex: 0.01
    N = int(N)
    totalThreads = sys.argv[2]
    totalThreads = int(totalThreads)
    print("N = ", N)
    print("totalThreads = ", totalThreads)
else:
    print("Incorrect number of arguments!")
    sys.exit(1)

if ((totalThreads == 1) or (totalThreads == 2) or (totalThreads == 4) or (totalThreads == 8)):
    print()
else:
    print("Invalid number of threads. Please use 1, 2, 4, or 8 threads.")
    sys.exit(1)

# start experiment
t = int(time() * 1000)  # begin run time
total = 0

# ATTEMPT 1
# processes = []
# for i in range(totalThreads):
#     process = mp.Process(target=myThread, args=(N/totalThreads))
#     processes.append(process)
#     process.start()
    
# for process in processes:
#     process.join()

# ATTEMPT 2
#pool = mp.Pool(mp.cpu_count())
#total = pool.map(myThread, [N/totalThreads])

# ATTEMPT 3
#for i in range(totalThreads):
    #total = total + pool.map(myThread, [N/totalThreads])
#    p = mp.Process(target=myThread, args=(N/totalThreads))
#    p.start()

# ATTEMPT 4
# with concurrent.futures.ThreadPoolExecutor() as executor:
#     for i in range(totalThreads):
#         future = executor.submit(myThread, N/totalThreads) # start thread
#         total = total + future.result() # get result

# analyze results
pi = 4 * total / N
print("pi estimate =", pi)
delta_time = int(time() * 1000) - t # calculate time required
print("Time =", delta_time, " milliseconds")

我认为创建一个从 0 到 totalThreads 的循环，为每次迭代创建一个进程是可行的。我还想传入 N/totalThreads（以划分工作），但似乎进程接受一个可迭代列表而不是传递给方法的参数。

多处理缺少什么？甚至有可能对流程做我想做的事情吗？ 提前感谢您的帮助，非常感谢:)

Answer 1

我已经简化了您的代码并使用了一些可能合理也可能不合理的硬编码值。

import math
import concurrent.futures
import random
from datetime import datetime


def myThread(arg):
    count = 0
    for i in range(0, arg[0]):
        x = random.uniform(0, 1)
        y = random.uniform(0, 1)
        d = math.sqrt(x * x + y * y)
        if (d < 1):
            count += 1
    return count


N = 10_000
T = 8

_start = datetime.now()
with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = {executor.submit(myThread, (int(N / T),)): _ for _ in range(T)}
    total = 0
    for future in concurrent.futures.as_completed(futures):
        total += future.result()
_end = datetime.now()
print(f'Estimate for PI = {4 * total / N}')
print(f'Run duration = {_end-_start}')

我机器上的典型输出如下所示：-

估计 PI = 3.1472
运行时长 = 0:00:00.008895

请记住，您启动的线程数由 ThreadPoolExecutor (TPE) [在没有参数构造时] 有效管理。它根据机器的处理能力（内核数量等）决定可以运行的线程数。因此，如果您真的愿意，您可以将 T 设置为一个非常高的数字，并且 TPE 将阻止任何新线程的执行，直到它确定有容量。