自定义Process类的Python进程池无法重启子进程问题

问题背景 #

基于multiprocess.Process重写了自定义Process类，单独使用时日志输出正常，但在进程池中设置maxtasksperchild=1时出现两个问题：无日志输出、进程重启失败报错。

自定义Process类代码 #

# source.Process.py
import multiprocess
import time
import traceback
from your_logger_module import S_Logger  # 替换为实际日志模块路径

class Process(multiprocess.Process):
    def __init__(self, test_name: str, *args, **kwargs) -> None:
        multiprocess.Process.__init__(self, *args, **kwargs)
        self._parent_conn, self._child_conn = multiprocess.Pipe()
        self._exception = None
        self._test_name = test_name

    def run(self) -> None:
        try:
            start = time.perf_counter()
            logger = S_Logger(self._test_name).get_logger()
            logger.info('EXECUTION OF %s HAS STARTED.', self._test_name)
            multiprocess.Process.run(self)
            self._child_conn.send(None)

        except Exception as e:
            tb = traceback.format_exc()
            logger.error(f'EXCEPTION OCCURRED: {tb}')
            self._child_conn.send((e, tb))

        finally:
            logger.info('EXECUTION OF %s HAS ENDED.', self._test_name)
            end = time.perf_counter()
            logger.info(f'FINISHED in {round(end-start, 2)} second(s)')

进程池代码 #

from source.process import Process
import multiprocess
import time

def run_tests(self):
    def worker(x):
        print(x**2)
        time.sleep(1)

    ctx = multiprocess.get_context()
    with ctx.Pool(processes=4, maxtasksperchild=1) as pool:
        nums = range(10)
        ctx.Process = Process(test_name='test_name')
        pool.map(worker, nums)

报错信息 #

0
1
4
9
Exception in thread Thread-1 (_handle_workers):
Traceback (most recent call last):
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.10_3.10.2032.0_x64__qbz5n2kfra8p0\lib\threading.py", line 1016, in _bootstrap_inner
    self.run()
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.10_3.10.2032.0_x64__qbz5n2kfra8p0\lib\threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\<user>\Documents\Projects\sprinter\.venv\lib\site-packages\multiprocess\pool.py", line 513, in _handle_workers
    cls._maintain_pool(ctx, Process, processes, pool, inqueue,
  File "C:\Users\<user>\Documents\Projects\sprinter\.venv\lib\site-packages\multiprocess\pool.py", line 337, in _maintain_pool
    Pool._repopulate_pool_static(ctx, Process, processes, pool,
  File "C:\Users\<user>\Documents\Projects\sprinter\.venv\lib\site-packages\multiprocess\pool.py", line 319, in _repopulate_pool_static
    w = Process(ctx, target=worker,
  File "C:\Users\<user>\Documents\Projects\sprinter\.venv\lib\site-packages\multiprocess\pool.py", line 181, in Process
    return ctx.Process(*args, **kwds)
TypeError: 'Process' object is not callable

疑问与解答 #

1. 为什么没有日志输出？ #

• 进程池初始化顺序错误：创建Pool之后才给ctx.Process赋值，此时初始4个工作进程已经用默认Process类创建，根本没用到自定义Process，自然不会触发自定义run方法里的日志逻辑。
• 后续进程重启失败，导致自定义Process从未被正常使用，进一步无日志输出。
• 额外注意：需确保S_Logger支持多进程场景（比如使用进程安全的日志handler，避免日志错乱）。

2. 为什么进程重启失败报TypeError: 'Process' object is not callable？ #

• 核心错误：ctx.Process = Process(test_name='test_name')是把Process类的实例赋值给了ctx.Process，而非类本身。进程池重建进程（因maxtasksperchild=1，每个进程仅执行1个任务）时，需要调用ctx.Process创建新进程，但此时ctx.Process是实例对象而非可调用的类，因此报错。
• 附加问题：自定义Process的__init__方法将test_name作为第一个必填参数，而进程池创建进程时不会传递该参数，即使赋值类也会因参数不匹配报错。

修正方案 #

步骤1：修正自定义Process类 #

调整__init__参数顺序，给test_name设置默认值，兼容进程池调用逻辑；同时增加日志初始化容错：

# source.Process.py
import multiprocess
import time
import traceback
from your_logger_module import S_Logger

class Process(multiprocess.Process):
    # 将test_name设为可选参数，放在*args之后
    def __init__(self, *args, test_name: str = "default_worker", **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self._parent_conn, self._child_conn = multiprocess.Pipe()
        self._exception = None
        self._test_name = test_name

    def run(self) -> None:
        logger = None
        try:
            start = time.perf_counter()
            logger = S_Logger(self._test_name).get_logger()
            logger.info('EXECUTION OF %s HAS STARTED.', self._test_name)
            super().run()
            self._child_conn.send(None)

        except Exception as e:
            tb = traceback.format_exc()
            if logger:
                logger.error(f'EXCEPTION OCCURRED: {tb}')
            self._child_conn.send((e, tb))

        finally:
            if logger:
                logger.info('EXECUTION OF %s HAS ENDED.', self._test_name)
                end = time.perf_counter()
                logger.info(f'FINISHED in {round(end-start, 2)} second(s)')

步骤2：修正进程池代码 #

在创建Pool之前设置ctx.Process为自定义类，而非实例：

from source.process import Process
import multiprocess
import time

def run_tests():
    def worker(x):
        print(x**2)
        time.sleep(1)

    ctx = multiprocess.get_context()
    # 赋值自定义Process类，而非实例
    ctx.Process = Process
    with ctx.Pool(processes=4, maxtasksperchild=1) as pool:
        nums = range(10)
        pool.map(worker, nums)

关键修正点总结 #

• 调整自定义Process的__init__参数，确保兼容进程池的默认调用逻辑。
• 给ctx.Process赋值类而非实例，让进程池能正常调用类创建新进程。
• 提前设置ctx.Process，确保初始工作进程也使用自定义Process，日志正常输出。
• 增加日志初始化容错，避免因日志模块问题导致额外错误。

内容的提问来源于stack exchange，提问作者Alraku