在 macOS 中进行多进程编程时，Python 可以使用 `multiprocessing` 模块结合信号量或超时参数来处理超时情况。以下是几种常见的处理方法。 ### 使用 `multiprocessing.Process` 和 `join` 的超时参数 ```python import multiprocessing import time def long_running_task(): time.sleep(10) print("Task completed") if __name__ == "__main__": p = multiprocessing.Process(target=long_running_task) p.start() # 设置超时时间为 5 秒 p.join(timeout=5) if p.is_alive(): print("Task timed out, terminating...") p.terminate() p.join() ``` 在这个示例中，`long_running_task` 函数模拟一个耗时 10 秒的任务。在主程序中，创建一个新的进程来运行这个任务，并通过 `join` 方法的 `timeout` 参数设置超时时间为 5 秒。如果超时后任务仍在运行，则使用 `terminate` 方法终止该进程。 ### 使用 `multiprocessing.Pool` 和 `apply_async` 的 `get` 超时参数 ```python import multiprocessing import time def long_running_task(): time.sleep(10) return "Task completed" if __name__ == "__main__": pool = multiprocessing.Pool(processes=1) result = pool.apply_async(long_running_task) try: # 设置超时时间为 5 秒 output = result.get(timeout=5) print(output) except multiprocessing.TimeoutError: print("Task timed out") pool.terminate() pool.join() ``` 这里使用 `multiprocessing.Pool` 创建一个进程池，并使用 `apply_async` 异步执行任务。通过 `get` 方法的 `timeout` 参数设置超时时间，如果超时则捕获 `TimeoutError` 异常并终止进程池。 ### 使用信号处理超时 ```python import multiprocessing import signal import time def long_running_task(): time.sleep(10) print("Task completed") def timeout_handler(signum, frame): raise TimeoutError("Task timed out") if __name__ == "__main__": p = multiprocessing.Process(target=long_running_task) p.start() signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(5) # 设置 5 秒的超时时间 try: p.join() except TimeoutError: print("Task timed out, terminating...") p.terminate() p.join() finally: signal.alarm(0) # 取消闹钟 ``` 此示例使用 `signal` 模块设置一个闹钟信号，当超时发生时触发 `timeout_handler` 函数，抛出 `TimeoutError` 异常，然后终止进程。