Python Selenium多进程执行后如何批量关闭所有WebDriver实例?
问题根源
多进程环境下内存是相互隔离的,你在子进程中往全局drivers列表添加的Driver实例,并不会同步到主进程的drivers列表里,所以主进程遍历列表执行driver.close()时,实际操作的是空列表,自然无法关闭浏览器。
可行解决方案
方案1:用进程事件同步,子进程自行关闭Driver
通过multiprocessing.Event实现主进程统一触发关闭信号,子进程收到信号后自行关闭Driver,避开跨进程传递实例的问题:
import multiprocessing as mp import time from selenium import webdriver # 导入你的初始化函数相关依赖 def initialize_driver(starting_url: str = 'https://www.google.com/'): chrome_options = webdriver.ChromeOptions() chrome_options.add_experimental_option("detach", True) driver = webdriver.Chrome( service=webdriver.chrome.service.Service(webdriver_manager.chrome.ChromeDriverManager().install()), options=chrome_options) driver.get(starting_url) driver.implicitly_wait(10) time.sleep(1) return driver def worker_bot_test(shutdown_event): driver = initialize_driver() # 执行你的任务逻辑 print("任务执行中...") time.sleep(5) # 等待主进程的关闭指令 shutdown_event.wait() # 彻底关闭Driver及对应Chrome进程 driver.quit() if __name__ == "__main__": shutdown_event = mp.Event() workers = [] # 启动子进程 for _ in range(4): worker = mp.Process(target=worker_bot_test, args=(shutdown_event,)) worker.start() workers.append(worker) # 确认所有子进程任务执行完成 # 这里可根据实际任务逻辑调整,比如用队列接收任务完成信号 time.sleep(6) # 触发统一关闭信号 shutdown_event.set() # 等待所有子进程完成关闭操作 for worker in workers: worker.join()
方案2:收集Chrome进程PID,主进程批量终止
每个Selenium Driver对应一个独立的Chrome进程,我们可以在子进程中获取进程PID,传递给主进程后批量杀死:
import multiprocessing as mp import psutil import time from selenium import webdriver def initialize_driver(starting_url: str = 'https://www.google.com/'): chrome_options = webdriver.ChromeOptions() chrome_options.add_experimental_option("detach", True) driver = webdriver.Chrome( service=webdriver.chrome.service.Service(webdriver_manager.chrome.ChromeDriverManager().install()), options=chrome_options) driver.get(starting_url) driver.implicitly_wait(10) time.sleep(1) # 获取对应Chrome进程的PID chrome_pid = driver.service.process.pid return driver, chrome_pid def worker_bot_test(pid_queue): driver, chrome_pid = initialize_driver() # 执行任务逻辑 print("任务执行中...") time.sleep(5) # 将PID发送到主进程的队列 pid_queue.put(chrome_pid) # 先关闭浏览器窗口,进程留待主进程终止 driver.close() if __name__ == "__main__": pid_queue = mp.Queue() workers = [] # 启动子进程 for _ in range(4): worker = mp.Process(target=worker_bot_test, args=(pid_queue,)) worker.start() workers.append(worker) # 等待所有子进程完成任务 for worker in workers: worker.join() # 批量终止Chrome进程 while not pid_queue.empty(): pid = pid_queue.get() try: process = psutil.Process(pid) process.terminate() process.wait() print(f"已关闭PID为{pid}的Chrome进程") except psutil.NoSuchProcess: print(f"PID为{pid}的进程已不存在")
方案3:禁用detach选项,Driver随子进程自动关闭
如果不需要Chrome在子进程结束后继续运行,直接移除detach配置,这样子进程执行完毕后,对应的Driver和Chrome进程会自动关闭:
def initialize_driver(starting_url: str = 'https://www.google.com/'): chrome_options = webdriver.ChromeOptions() # 移除detach配置 # chrome_options.add_experimental_option("detach", True) driver = webdriver.Chrome( service=webdriver.chrome.service.Service(webdriver_manager.chrome.ChromeDriverManager().install()), options=chrome_options) driver.get(starting_url) driver.implicitly_wait(10) time.sleep(1) return driver
主进程原有的多进程代码无需修改,子进程任务完成后,Driver会随子进程结束自动关闭。
内容的提问来源于stack exchange,提问作者jsv
相关产品推荐
相关产品推荐

