Python Selenium：关闭 webdriver 的所有实例

Question

我正在研究这个浏览器自动化项目，它并行执行一些浏览器任务。这个想法是：

• 打开四个浏览器
• 做一些任务
• 在我们关闭所有浏览器之前等待所有浏览器完成任务

这是一个用于演示目的的简单 Web 驱动程序函数。

# For initializing webdriver
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.options import Options

def initialize_driver(starting_url: str = 'https://www.google.com/'):
    ''' Open a webdriver and go to Google
    '''
    # Webdriver option(s): keep webdriver opened
    chrome_options = Options()
    chrome_options.add_experimental_option("detach", True) 

    # Initialize webdriver
    driver = webdriver.Chrome(
         service=Service(ChromeDriverManager().install()), 
         options=chrome_options)
    
    # Open website; wait until fully loaded
    driver.get(starting_url)
    driver.implicitly_wait(10)
    time.sleep(1)

    return driver

使用这个函数，我现在可以使用multiprocessing 创建四个并行运行的作业。

# Import package
import multiprocessing as mp

# List of workers
workers = []

# Run in parallel
for _ in range(4):
    worker = mp.Process(target=phm2.worker_bot_test)
    worker.start()
    workers.append(worker)

for worker in workers:
    worker.join()

这些已经涵盖了前两点，但据我所知，我们一次只能使用driver.close() 关闭一个 webdriver。有没有一种方法可以一次性关闭它们？我实际上尝试创建一个 webdrivers 列表并在函数末尾附加一个 webdriver。然后，一一关闭。但由于某种原因，它不起作用。

# I added drivers.append(driver) at the end of the function from earlier
# This will now be a global variable to store the list of drivers
drivers = []

# Insert multiprocessing code here...

# Close all drivers
for driver in drivers:
   driver.close()

我可以尝试做些什么来完成最后一步？我一直看到我们可以调整Process 类以包含返回值（有返回值将是一个很大的帮助），但是，我尽可能不想这样做，因为它有点复杂。

Answer 1

每个webdriver 对象都是绝对独立的对象实例。
与您申请 f.e. 时的方式相同。 get() 某些特定 webdriver 对象上的方法，这对任何其他 webdriver 对象没有影响，同样，当您在某些 webdriver 对象上应用 quit() 或 close() 时，这绝对不会影响任何其他 @987654328 @ 目的。
因此，关闭所有webdriver 会话的唯一方法是将所有webdriver 对象保留在某种结构中，例如list 等。
当您需要关闭所有会话时，您需要遍历该列表并将driver.quit() 应用于该列表中的每个对象。
顺便说一句，为了清楚地关闭会话，您应该使用quit() 方法，而不是close()。

Answer 2

我首先会注意到，由于 selenium 驱动程序已经作为子进程运行，因此您只需要使用多线程即可。我假设您的线程在检索网页及其元素后所做的任何工作都不是特别占用 CPU 资源的。如果不是这种情况，您始终可以创建一个多处理池，将其传递给worker_bot_test 工作函数，以并行执行任何 CPU 密集型操作。

通过使用线程，我们可以创建一个创建驱动程序的类，并具有一个__del__ 终结器，当类实例被垃圾收集时，它会“退出”驱动程序。我们在线程本地存储中保留对该类实例的引用，以便仅在线程终止并且线程本地存储被垃圾收集时才调用终结器。为了确保这种垃圾回收，我们可以在子线程终止后显式调用gc.collect。如果我们使用多处理而不是多线程，对gc.collect 的调用将无效，因为它只会对当前进程进行垃圾收集。

# For initializing webdriver
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.options import Options

import threading

class ChromeDriver:
    def __init__(self, starting_url):
        chrome_options = Options()
        chrome_options.add_experimental_option("detach", True)
        # Not a bad option to add:
        #chrome_options.add_experimental_option('excludeSwitches', ['enable-logging'])
        # If we don't need to see the browsers:
        #chrome_options.add_argument("headless")

        # Initialize webdriver
        self.driver = webdriver.Chrome(
             service=Service(ChromeDriverManager().install()),
             options=chrome_options)

        # Open website; wait until fully loaded
        self.driver.get(starting_url)
        self.driver.implicitly_wait(10)
        # What is the purpose of the following line?
        #time.sleep(1)

    def __del__(self):
        self.driver.quit() # clean up driver when we are cleaned up
        print('The driver has been "quitted".')

threadLocal = threading.local()

def initialize_driver(starting_url: str = 'https://www.google.com/'):
    chrome_driver =  ChromeDriver(starting_url)
    # Make sure there is a reference to the ChromeDriver instance so that
    # it is not prematurely finalized:
    threadLocal.driver = chrome_driver
    return chrome_driver.driver

def worker_bot_test():
    driver = initialize_driver()
    print(len(driver.page_source))


if __name__ == '__main__':
    # List of workers
    workers = []

    # Run in parallel
    for _ in range(4):
        worker = threading.Thread(target=worker_bot_test)
        worker.start()
        workers.append(worker)

    for worker in workers:
        worker.join()

    # Ensure finalizers are executed:
    import gc
    gc.collect()

印刷：

...
163036
163050
163183
165486
The driver has been "quitted".
The driver has been "quitted".
The driver has been "quitted".
The driver has been "quitted".