如果运行时间超过 x 分钟，则重新启动进程

Question

我有一个程序可以创建一个多处理池来处理 Web 提取作业。本质上，产品 ID 的列表被输入到一个由 10 个处理队列的进程组成的池中。代码很简单：

import multiprocessing

num_procs = 10

products = ['92765937', '20284759', '92302047', '20385473', ...etc]

def worker():
    for workeritem in iter(q.get, None):
        time.sleep(10)
        get_product_data(workeritem)
        q.task_done()
    q.task_done()

q = multiprocessing.JoinableQueue()

procs = []

for i in range(num_procs):
    procs.append(multiprocessing.Process(target=worker))
    procs[-1].daemon = True
    procs[-1].start()

for product in products:
    time.sleep(10)
    q.put(product)

q.join()

for p in procs:
    q.put(None)

q.join()

for p in procs:
    p.join()

get_product_data() 函数获取产品，打开 Selenium 实例，导航到站点，登录并收集产品详细信息并输出到 csv 文件。问题是，随机（字面意思......它发生在网站导航或提取过程的不同点）Selenium 将停止做它正在做的任何事情，只是坐在那里停止做它的工作。不会抛出异常或任何东西。我已经在get_product_data() 函数中尽我所能避免这种情况发生，但这似乎只是 Selenium 的一个问题（我尝试使用 Firefox、PhantomJS 和 Chrome 作为它的驱动程序，但仍然遇到无论如何，同样的问题）。

基本上，该过程的运行时间不应超过 10 分钟。如果进程运行时间超过指定时间，是否有任何方法可以杀死进程并使用相同的产品 ID 重新启动它？

这一切都在使用 Python 2.7 的 Debian Wheezy 机器上运行。

Answer 1

您可以使用multiprocessing.Pool 和timeout() function suggested by @VooDooNOFX 编写代码。未经测试，将其视为可执行伪代码：

#!/usr/bin/env python
import signal
from contextlib import closing
from multiprocessing import Pool

class Alarm(Exception):
    pass

def alarm_handler(*args):
    raise Alarm("timeout")

def mp_get_product_data(id, timeout=10, nretries=3):
    signal.signal(signal.SIGALRM, alarm_handler) #XXX could move it to initializer
    for i in range(nretries):
        signal.alarm(timeout) 
        try:
            return id, get_product_data(id), None
        except Alarm as e:
            timeout *= 2 # retry with increased timeout
        except Exception as e:
            break
        finally:
            signal.alarm(0) # disable alarm, no need to restore handler
    return id, None, str(e) 

if __name__=="__main__":
   with closing(Pool(num_procs)) as pool:
       for id, result, error in pool.imap_unordered(mp_get_product_data, products):
           if error is not None: # report and/or reschedule
              print("error: {} for {}".format(error, id))
   pool.join()

Answer 2

你需要让 Selenium 等待一段明确的时间，或者等待一些隐式的 DOM 对象可用。 Take a quick look at the selenium docs about that.

从链接来看，这是一个等待10 seconds DOM 元素myDynamicElement 出现的过程。

from selenium import webdriver
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait # available since 2.4.0
from selenium.webdriver.support import expected_conditions as EC # available since 2.26.0

ff = webdriver.Firefox()
ff.get("http://somedomain/url_that_delays_loading")
try:
    element = WebDriverWait(ff, 10).until(EC.presence_of_element_located((By.ID, "myDynamicElement")))
except TimeoutException as why:
    # Do something to reject this item, possibly by re-adding it to the worker queue.
finally:
    ff.quit()

如果在给定时间段内没有可用的内容，则会引发 selenium.common.exceptions.TimeoutException，您可以像上面一样在 try/except 循环中捕获它。

编辑

另一种选择是让multiprocessing 在一段时间后使进程超时。这是使用内置库signal 完成的。 Here's an excellent example of doing this，但是当您检测到进程已被终止时，仍然需要将该项目添加回工作队列。您可以在代码的def handler 部分执行此操作。