如何用Python多进程池调用跨类/模块方法并实现任务等待机制

没问题，我来给你梳理一个落地的解决方案——你需要的是一个常驻等待任务的进程池，能调用不同模块类的方法，完成后回到等待状态。核心思路是用concurrent.futures.ProcessPoolExecutor（对应你说的ProcessExecutor）搭配跨进程任务队列来实现循环等待逻辑，下面结合你提到的三个模块结构来写具体代码：

1. 任务模块：Reader类所在的模块（比如reader_module.py） #

这里是你的任务执行类，包含实际要运行的业务逻辑：

# reader_module.py
class Reader:
    def __init__(self, task_detail):
        self.task_detail = task_detail

    def process_task(self):
        # 替换成你的实际任务逻辑：比如读取文件、处理数据等
        print(f"开始处理任务: {self.task_detail}")
        # 模拟任务耗时
        import time
        time.sleep(2)
        return f"任务[{self.task_detail}]执行完成"

2. 进程池管理模块：ProcessExecutor（比如process_executor.py） #

这个类负责管理进程池、维护任务队列，实现持续等待/执行的循环：

# process_executor.py
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager
import threading
import time

class ProcessExecutor:
    def __init__(self, max_workers=3):
        # 用Manager创建跨进程通信的任务队列（普通queue.Queue无法跨进程使用）
        self.task_queue = Manager().Queue()
        # 初始化进程池
        self.pool = ProcessPoolExecutor(max_workers=max_workers)
        # 控制循环运行的标志
        self.is_running = True

    def _task_loop(self):
        """内部循环：持续从队列取任务并提交到进程池"""
        while self.is_running:
            try:
                # 阻塞等待任务，超时1秒用于检查是否停止运行
                task_payload = self.task_queue.get(timeout=1)
                # 解析任务：可调用对象 + 位置参数 + 关键字参数
                callable_func, args, kwargs = task_payload
                # 异步提交任务到进程池
                future = self.pool.submit(callable_func, *args, **kwargs)
                # 添加回调函数处理任务结果（可选，比如记录日志、通知主进程）
                future.add_done_callback(self._handle_task_result)
            except Exception:
                # 超时异常是正常的，用于退出循环前的检查
                if not self.is_running:
                    break

    def _handle_task_result(self, future):
        """处理任务完成后的结果/异常"""
        try:
            result = future.result()
            print(f"任务结果: {result}")
        except Exception as e:
            print(f"任务执行失败: {str(e)}")

    def submit_task(self, func, *args, **kwargs):
        """外部提交任务的接口：接收可调用对象和参数"""
        self.task_queue.put((func, args, kwargs))

    def start(self):
        """启动进程池和任务循环"""
        # 用线程运行任务循环，避免阻塞主进程
        threading.Thread(target=self._task_loop, daemon=True).start()
        print("进程池已启动，等待任务中...")

    def stop(self):
        """停止进程池和循环"""
        self.is_running = False
        # 等待所有已提交的任务完成后关闭进程池
        self.pool.shutdown(wait=True)
        print("进程池已停止")

3. 主模块：启动进程池并提交任务（比如main.py） #

这里是程序的入口，负责初始化进程池、模拟任务提交（实际场景中可以是从API、消息队列等接收任务）：

# main.py
from process_executor import ProcessExecutor
from reader_module import Reader
import time

def run_reader_task(task_id):
    """封装Reader类的调用：因为多进程中类实例需要可序列化，封装成普通函数更稳妥"""
    reader = Reader(f"用户任务-{task_id}")
    return reader.process_task()

if __name__ == "__main__":
    # 初始化进程池（设置最大工作进程数）
    executor = ProcessExecutor(max_workers=2)
    executor.start()

    try:
        # 模拟外部任务提交（实际场景可替换为从消息队列/API接收任务）
        time.sleep(2)
        print("提交第一个任务")
        executor.submit_task(run_reader_task, 1)

        time.sleep(3)
        print("提交第二个任务")
        executor.submit_task(run_reader_task, 2)

        # 让主进程持续运行，等待用户中断
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n收到中断信号，正在停止进程池...")
        executor.stop()

关键要点说明 #

• 跨进程队列的选择：必须用multiprocessing.Manager().Queue()，普通的queue.Queue只能在单进程内使用，无法实现主进程和子进程的任务传递。
• 任务封装的技巧：直接传递类的方法可能会遇到序列化问题（如果类包含不可序列化的属性，比如文件句柄），所以把类实例化和方法调用封装成普通函数（比如示例中的run_reader_task）是更稳妥的方案。
• 常驻循环的实现：用线程来运行任务循环，这样主进程可以同时处理其他逻辑（比如监听外部任务请求），不会被阻塞。
• 进程池的优雅关闭：调用shutdown(wait=True)可以等待所有已提交的任务完成后再关闭进程池，避免任务丢失或资源泄漏。

额外注意事项 #

• 如果你的Reader类包含不可序列化的属性，一定要在子进程内部完成实例化（就像示例中的run_reader_task那样），不要在主进程实例化后传递给子进程。
• 若需要更可靠的任务管理（比如任务持久化、失败重试），可以考虑搭配第三方消息队列（如Redis、RabbitMQ）替代Manager().Queue()，但简单场景下后者足够使用。
• 如果你更习惯用multiprocessing.Pool而非ProcessPoolExecutor，逻辑是类似的：用Pool.apply_async()异步提交任务，配合队列循环即可。

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