为什么 ThreadPool 在 TimeoutError 之后不移动答案

【问题标题】：Why ThreadPool doesn't move next after a TimeoutError为什么 ThreadPool 在 TimeoutError 之后不移动
【发布时间】：2014-02-03 09:58:48
【问题描述】：

我的目标：

使用请求浏览网站列表以检查它们。这是在apply_job 中完成的。

我的问题：

当调用job_pool.next 时，一些网站出现错误，而不是给出错误，他们只是站在那里甚至不给出TimeoutError。这就是为什么我在 next 函数中使用超时时间为 10 秒的原因。此超时运行良好，但当出现TimeoutError 异常时，next 函数在以下时间继续引发异常，即使下一个网站很好。在我看来，它并没有移动到下一个项目，而是在同一个项目上循环。
我试过imap 和imap_unordered，没有区别。

我的代码在这里：

   def run_check(websites):
        """ Run check on the given websites """
        import multiprocessing
        from multiprocessing.pool import ThreadPool

        pool = ThreadPool(processes=JOB_POOL_SIZE)

        try:
            job_pool = pool.imap_unordered(apply_job, websites)

            try:
                while True:
                    try:
                        res = job_pool.next(10)
                    except multiprocessing.TimeoutError:
                        logging.error("Timeout Error")
                        res = 'No Res'

                    csv_callback(res)

            except StopIteration:
                pass

            pool.terminate()
        except Exception, e:
            logging.error("Run_check Error: %s"%e)
            raise

我使用res = requests.get(url, timeout=10) 来查看网站。此超时不适用于此问题。

为了测试，这里是出现问题的网站（不是每次，但经常出现）：http://www.kddecorators.netfirms.com、http://www.railcar.netfirms.com。

我不知道这些网站有什么不同，但我的猜测是它们会不时发送一个字节，因此即使它们不可用，也不会被视为真正的超时。

如果有人有想法，将不胜感激，我已经坚持了几天了。我什至尝试过future 和async，但它们没有引发我需要的异常。

谢谢大家！

【问题讨论】：

标签： python multithreading timeout threadpool

【解决方案1】：

您认为将超时传递给next 会中止工作的直觉是错误的。它只是中止等待，但特定作业继续运行。下次你等待时，你确实在等待同样的工作。要实现实际作业的超时，您应该查看requests documentation。请注意，没有可靠的方法来终止另一个线程。因此，如果您绝对不能让您的作业在合理的时间范围内终止，您可以切换到基于进程的池并强制终止进程（例如使用signal.alarm）。

【讨论】：

【解决方案2】：

我找到了解决问题的方法，我使用了eventlet 及其Timeout 函数。

def apply_job(account_info):
    """ Job for the Thread """
    try:
        account_id = account_info['id']
        account_website = account_info['website']
        url = account_website
        result = "ERROR: GreenPool Timeout"
        with Timeout(TIMEOUT*2, False):
            url, result = tpool.execute(website.try_url, account_website)

        return (account_id, account_website, url, result)

    except Exception, e:
        logging.error("Apply_job Error: %s"%e)

def start_db(res):
    update_db(res)
    csv_file.csv_callback(res)

def spawn_callback(result):
    res = result.wait()
    tpool.execute(start_db, res)

def run_check(websites):
    """ Run check on the given websites """
    print str(len(websites)) + " items found\n"

    pool = eventlet.GreenPool(100)
    for i, account_website in enumerate(websites):
        res = pool.spawn(apply_job, account_website)
        res.link(spawn_callback)

    pool.waitall()

此解决方案运行良好，因为它在命令 url, result = tpool.execute(website.try_url, account_website) 中函数 website.try_url 的整个执行过程中超时。

【讨论】：