如何在终止多进程时关闭Selenium ChromeDriver进程
解决多进程终止时ChromeDriver残留的问题
直接调用terminate()会强制终止进程,进程内的ChromeDriver和Chrome实例根本没机会执行关闭逻辑,所以会残留后台进程。下面给你几个可行的解决思路:
方案1:用事件通知进程主动退出
给每个子进程传递一个multiprocessing.Event对象,主进程需要终止时触发事件,子进程检测到信号后主动关闭Driver再退出:
import multiprocessing from time import sleep from selenium import webdriver def start_process(url, stop_event): driver = webdriver.Chrome() try: driver.get(url) # 业务逻辑执行期间循环检测终止信号 while not stop_event.is_set(): sleep(1) finally: driver.quit() # 确保Driver被正常关闭 if __name__ == "__main__": urls = ["https://example.com"] * 2 # 示例URL列表 stop_events = [] processes = [] for counter, url in enumerate(urls): stop_event = multiprocessing.Event() proc = multiprocessing.Process( target=start_process, name=f'start_process_{counter}', args=(url, stop_event) ) processes.append(proc) stop_events.append(stop_event) proc.start() sleep(10) # 先触发终止事件,让进程主动清理资源 for event in stop_events: event.set() # 等待进程退出,超时再强制终止 for proc in processes: proc.join(timeout=5) if proc.is_alive(): proc.terminate()
方案2:捕获终止信号,在信号处理中关闭Driver
给子进程注册SIGTERM信号处理函数,收到终止信号时主动关闭Driver:
import multiprocessing import signal from time import sleep from selenium import webdriver def handle_termination(signum, frame, driver): driver.quit() exit(0) def start_process(url): driver = webdriver.Chrome() # 绑定信号处理函数,传入当前Driver实例 signal.signal(signal.SIGTERM, lambda s, f: handle_termination(s, f, driver)) driver.get(url) # 模拟业务持续运行 while True: sleep(1) if __name__ == "__main__": urls = ["https://example.com"] * 2 processes = [] for counter, url in enumerate(urls): proc = multiprocessing.Process( target=start_process, name=f'start_process_{counter}', args=(url,) ) processes.append(proc) proc.start() sleep(10) for proc in processes: proc.terminate() proc.join()
方案3:强制清理残留进程(兜底方案)
如果上面的方法都不适用,可以在主进程终止子进程后,手动查找并杀死相关的ChromeDriver/Chrome进程(注意:该方法可能误杀无关进程,谨慎使用):
import multiprocessing import os from time import sleep from selenium import webdriver def start_process(url): driver = webdriver.Chrome() driver.get(url) while True: sleep(1) if __name__ == "__main__": urls = ["https://example.com"] * 2 processes = [] for counter, url in enumerate(urls): proc = multiprocessing.Process( target=start_process, name=f'start_process_{counter}', args=(url,) ) processes.append(proc) proc.start() sleep(10) for proc in processes: proc.terminate() proc.join() # Linux/macOS下清理进程,Windows需替换为taskkill命令 os.system("pkill -f chromedriver") os.system("pkill -f chrome")
内容的提问来源于stack exchange,提问作者user2396640
相关产品推荐
相关产品推荐

