我已经定义了这个函数

def writeonfiles(a,seed):
    random.seed(seed)

    f = open(a, "w+")
    for i in range(0,10):
        j = random.randint(0,10)
        #print j
        f.write(j)
    f.close()

其中 a 是包含文件路径的字符串，seed 是整数种子。我想以这样的方式并行化一个简单的程序，即每个核心采用我提供的可用路径之一，为其随机生成器提供种子并在该文件上写入一些随机数，例如，如果我传递向量

vector = [Test/file1.txt, Test/file2.txt] 

和种子

seeds = (123412, 989898), 

它为第一个可用的核心提供了功能

writeonfiles(Test/file1.txt, 123412) 

和第二个具有不同参数的相同函数：

writeonfiles(Test/file2.txt, 989898)

我在 Stackoverflow 上浏览了很多类似的问题，但我无法做出任何解决方案。我试过的是：

def writeonfiles_unpack(args):
    return writeonfiles(*args)
if __name__ == "__main__":
     folder = ["Test/%d.csv" %i for i in range(0,4)]
     seed = [234124, 663123, 12345 ,123833]
     p = multiprocessing.Pool()
     p.map(writeonfiles, (folder,seed))

并给我 TypeError: writeonfiles() 正好需要 2 个参数（给定 1 个）。

我也试过

if __name__ == …

考虑以下代码示例：

import multiprocessing
import requests

session = requests.Session()
data_to_be_processed = [...]

def process(arg):
    # do stuff with arg and get url
    response = session.get(url)
    # process response and generate data...
    return data

with multiprocessing.Pool() as pool:
    results = pool.map(process, data_to_be_processed)

例如，Session被分配为全局变量，因此在其中创建进程Pool后将被复制到每个子进程中。我不确定会话是否是线程安全的，也不知道会话中的池是如何工作的，所以我想为 pool 中的每个进程分配单独的会话对象。

我知道，我可以使用requests.get(url)而不是session.get(url)，但我想使用 session 并且我也在考虑使用requests-html（https://html.python-requests.org/）。

我对python的多处理不是很熟悉，到目前为止我只使用了池，因为它是我认为并行处理数据而没有临界区的最佳解决方案，所以我对不同的解决方案持开放态度。

有没有办法做到干净和直接？

类似的问题（但答案对我不起作用）：如何取消使用 concurrent.futures.ProcessPoolExecutor 运行的长时间运行的子进程？

与上面链接的问题和提供的解决方案不同，在我的情况下，计算本身相当长（受 CPU 限制）并且无法循环运行以检查是否发生了某些事件。

以下代码的简化版本：

import asyncio
import concurrent.futures as futures
import time

class Simulator:
    def __init__(self):
        self._loop = None
        self._lmz_executor = None
        self._tasks = []
        self._max_execution_time = time.monotonic() + 60
        self._long_running_tasks = []

    def initialise(self):
        # Initialise the main asyncio loop
        self._loop = asyncio.get_event_loop()
        self._loop.set_default_executor(
            futures.ThreadPoolExecutor(max_workers=3))

        # Run separate processes of long computation task
        self._lmz_executor = futures.ProcessPoolExecutor(max_workers=3)

    def run(self):
        self._tasks.extend(
            [self.bot_reasoning_loop(bot_id) for bot_id in [1, 2, 3]]
        )

        try:
            # Gather bot reasoner tasks
            _reasoner_tasks = …

这个来自 PYMOTW 的例子给出了一个例子，multiprocessing.Pool()其中processes传递的参数（工作进程数）是机器内核数的两倍。

pool_size = multiprocessing.cpu_count() * 2

（否则该类将默认为 just cpu_count()。）

这有什么道理吗？创建比核心数更多的工人有什么影响？是否有理由这样做，或者它可能会在错误的方向上施加额外的开销？我很好奇为什么它会一直包含在我认为是信誉良好的网站的示例中。

在最初的测试中，它实际上似乎会减慢速度：

$ python -m timeit -n 25 -r 3 'import double_cpus; double_cpus.main()'
25 loops, best of 3: 266 msec per loop
$ python -m timeit -n 25 -r 3 'import default_cpus; default_cpus.main()'
25 loops, best of 3: 226 msec per loop

double_cpus.py：

import multiprocessing

def do_calculation(n):
    for i in range(n):
        i ** 2

def main():
    with multiprocessing.Pool(
        processes=multiprocessing.cpu_count() * 2,
        maxtasksperchild=2, …

我想使用multiprocessing.Value+multiprocessing.Lock在不同的进程之间共享一个计数器。例如：

import itertools as it
import multiprocessing

def func(x, val, lock):
    for i in range(x):
        i ** 2
    with lock:
        val.value += 1
        print('counter incremented to:', val.value)

if __name__ == '__main__':
    v = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()

    with multiprocessing.Pool() as pool:
        pool.starmap(func, ((i, v, lock) for i in range(25)))
    print(counter.value())

这将引发以下异常：

RuntimeError：同步对象只能通过继承在进程之间共享

我最困惑的是，一个相关的（虽然不是完全类似的）模式适用于multiprocessing.Process()：

if __name__ == '__main__':
    v = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()

    procs = [multiprocessing.Process(target=func, args=(i, v, lock))
             for i in …

1)multiprocessing模块是否支持可用于启动第二个进程而不是函数的 Python 脚本文件？

目前我使用multiprocessing.Processwhich 需要一个函数，但我想改为执行foo.py。我可以使用，subprocess.Popen但好处multiprocessing.Process是我可以传递对象（即使它们只是被腌制过）。

当我使用 multiprocessing.Process 时，为什么在子进程中导入了 my_module 而没有执行 print("foo") ？

2）当我使用 时multiprocessing.Process，为什么my_module在子进程中导入但print("foo")没有执行？尽管未执行主作用域，但 my_module 如何可用？

import multiprocessing
import my_module
print("foo")

def worker():
    print("bar")
    my_module.foo()
    return

p = multiprocessing.Process(target=worker, args=(1,2, d))
p.start()
p.join()

我正在对我拥有的数据集运行拼写校正功能。我曾经from pathos.multiprocessing import ProcessingPool as Pool做过这项工作。处理完成后，我想实际访问结果。这是我的代码：

import codecs
import nltk

from textblob import TextBlob
from nltk.tokenize import sent_tokenize
from pathos.multiprocessing import ProcessingPool as Pool

class SpellCorrect():

    def load_data(self, path_1):
        with codecs.open(path_1, "r", "utf-8") as file:
            data = file.read()
        return sent_tokenize(data)

    def correct_spelling(self, data):
        data = TextBlob(data)
        return str(data.correct())

    def run_clean(self, path_1):
        pool = Pool()
        data = self.load_data(path_1)
        return pool.amap(self.correct_spelling, data)

if __name__ == "__main__":
    path_1 = "../Data/training_data/training_corpus.txt"
    SpellCorrect = SpellCorrect()
    result = SpellCorrect.run_clean(path_1)
    print(result)
    result = " ".join(temp …

假设我有一组 20 个 CPU 繁重的任务（每个任务约 1 小时，但有些需要更长的时间），它们是通过调用函数来运行的，例如Pool.apply_async(function, task_list)等。PC 有 12 个内核。所以我可以分散负载，并使用所有 12 个内核。

每个任务的结果可能需要运行一个新任务（某些任务可能需要 1 次新运行，其他可能需要 10 次）。

当需要新任务时，我想将该任务生成到现有池中task_list，以始终充分优化 CPU 使用率。

目前我运行了 20 个任务，等待完成，启动新的 ~18 个任务，等待完成，启动剩余的新任务，......，而它恰好只有 1 个核心被使用了一个小时，而不是 12 个。这导致计算时间损失了几小时到几天。我可以在同一个工作人员中运行更新的任务，但这会导致更大的损失）

使用池似乎不可能在池启动时向池中添加更多任务。这是正确的，还是有一些聪明的方法可以做到这一点，而我在四处搜索时却错过了？

（我看到的唯一选项是使用process而不是pool，并在动态列表上进行 while 循环，该列表将典型任务作为单个进程启动，同时只允许最多 12 个进程同时运行，并且对于每个任务，或新的任务，将它们放在动态列表中，并在将任务发送到进程时将其删除。）

我有一个本质上只是一个无限循环的过程，我有第二个过程是一个计时器。计时器完成后如何终止循环进程？

def action():
  x = 0
  while True:
      if x < 1000000:
          x = x + 1
      else:
          x = 0

def timer(time):
  time.sleep(time)
  exit()    

loop_process = multiprocessing.Process(target=action)
loop_process.start()
timer_process = multiprocessing.Process(target=timer, args=(time,))
timer_process.start()

我希望 python 脚本在计时器完成后结束。

给定一个list()正在运行multiprocessing.Process的Process.join实例，我如何加入所有实例并在没有超时和循环的情况下在一个退出时立即返回？

例子

from multiprocessing import Process
from random import randint
from time import sleep
def run():
    sleep(randint(0,5))
running = [ Process(target=run) for i in range(10) ]

for p in running:
    p.start()

我怎样才能阻止，直到至少一个Process在p退出？

我不想做的是：

exit = False
while not exit:
    for p in running:
        p.join(0)
        if p.exitcode is not None:
            exit = True
            break