标签: python-multiprocessing

我有一个利用多处理来提高效率的 python 程序，以及一个为每个进程创建一个记录器的函数。记录器函数如下所示：

import logging
import os

def create_logger(app_name):
    """Create a logging interface"""
    # create a logger
    if logging in os.environ:
        logging_string = os.environ["logging"]
        if logging_string == "DEBUG":
            logging_level = loggin.DEBUG
        else if logging_string == "INFO":
            logging_level = logging.INFO
        else if logging_string == "WARNING":
            logging_level = logging.WARNING
        else if logging_string == "ERROR":
            logging_level = logging.ERROR
        else if logging_string == "CRITICAL":
            logging_level = logging.CRITICAL
    else:
        logging_level = logging.INFO

    logger = logging.getLogger(app_name)
    logger.setLevel(logging_level)

    # Console handler for error output
    console_handler = logging.StreamHandler() …

我有一个（大）名单，里面有男性和女性特工。

我想对每个应用不同的功能。

在这种情况下如何使用 Pool ？鉴于代理是相互独立的。

一个例子是：

males = ['a', 'b', 'c']
females = ['d', 'e', 'f']
for m in males:
    func_m(m)
for f in females:
    func_f(f)

我是这样开始的：

from multiprocessing import Pool
p = Pool(processes=2)
p.map() # Here is the problem

我想要一些类似的东西：

p.ZIP(func_f for f in females, func_m for m in males) # pseudocode

考虑以下代码示例：

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