标签: python-multiprocessing

考虑以下代码示例：

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 脚本在计时器完成后结束。

我创建了一个简单的脚本，它遍历我需要 ping 和 nslookup 的服务器列表。问题是，ping 可能需要一些时间，尤其是 ping 比一天中的秒数更多的服务器。

我对编程相当陌生，我知道多处理或多线程可能是使我的工作运行得更快的解决方案。

我的计划是获取我的服务器列表和 1. 将它分成大小均匀的列表，列表的数量与线程/进程匹配，或者 2. 如果这些选项之一支持它，则遍历单个列表，每个列表都传递一个新的在完成之前的 ping 和 nslookup 后，线程或进程的服务器名称。这是可取的，因为它确保我花费最少的时间，就好像列表 1 有 200 个离线服务器而列表 6 有 2000 个，它需要等待使用列表 6 的过程完成，即使所有其他人在那个时候都是免费的观点。

1. 哪一个更适合这项任务，为什么？

2. 如果可能，我将如何确保每个线程或进程具有基本相同的运行时

代码片段即使现在相当简单

import subprocess
import time
server_file = open(r"myfilepath", "r")
initial_time = time.time()
for i in range(1000):
    print(server_file.readline()[0:-1]+ ' '+str(subprocess.run('ping '+server_file.readline()[0:-1]).returncode)) #This returns a string with the server name, and return code,
print(time.time()-initial_time)

出现此问题是因为每次失败的 ping 平均需要 3 秒以上。此外，我知道不放置打印语句会使它更快，但我想监视它的一个小案例。我正在对 100,000 台服务器的影响进行 ping 操作，这将需要定期执行，并且该列表将继续增长

我正在尝试配置 Python 日志记录框架来处理来自multiprocessing.Pool. 在大多数情况下，脚本将无限期挂起，尽管已经观察到一些其他行为，例如退出而不打印所有日志消息。

我的实际代码更复杂，但我已将其简化为以下脚本，该脚本在我测试过的计算机上相当可靠地中断。

#!/usr/bin/env python3

import logging
import logging.handlers
import multiprocessing
import multiprocessing.util

L = logging.getLogger(__name__)

_globalQueue = None
_globalListener = None

def basicsetup():
    global _globalQueue
    global _globalListener

    cf = logging.Formatter("[{levelname}] {created:.7f} {name} ({process}~{processName}): {message}", style="{")

    handler = logging.StreamHandler()
    handler.setLevel(logging.DEBUG)
    handler.setFormatter(cf)

    # Subprocesses should use the queue to send log messages back to a thread in the main process
    _globalQueue = multiprocessing.Queue()

    _globalListener = logging.handlers.QueueListener(_globalQueue, handler, respect_handler_level=True)
    _globalListener.start()

    # Configure logging for main thread
    process_setup(get_queue())

def …