使用 ThreadPoolExecutor 强制线程超时

Question

我正在升级我的代码以使用 ThreadPoolExecuter，并希望能够使处理时间超过几秒钟的任何线程超时。是否可以对属于线程池的线程强制超时？我正在使用的代码如下。

    with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:
        future_tasks = {executor.submit(self.crawl_task, url): url for url in self.results.keys()}

        for future in concurrent.futures.as_completed(future_tasks):
            url = future_tasks[future]
            try:
                result = future.result()
                self.results[result[0]] = result[1]
            except Exception as e:
                print('%r generated an exception: %s' % (url, e))

我能够使线程超时的唯一方法是更改​​

for future in concurrent.futures.as_completed(future_tasks):

到

for future in concurrent.futures.as_completed(future_tasks, timeout=1):

但是，这会破坏整个循环，我将无法知道哪个线程超时以及哪些数据导致了超时。

Traceback (most recent call last):
  File "test.py", line 75, in <module>
    request = Requests(data)
  File "test.py", line 22, in __init__
    for future in concurrent.futures.as_completed(future_tasks, timeout=1):
  File "/source/homebrew/Cellar/python3/3.4.0_1/Frameworks/Python.framework/Versions/3.4/    lib/python3.4/concurrent/futures/_base.py", line 213, in as_completed
    len(pending), len(fs)))
concurrent.futures._base.TimeoutError: 17 (of 17) futures unfinished

Answer 1

将整个 for 循环包装在一个异常中仍然允许其他线程结果处理。使用两个单独的字典，您可以查看哪些线程因超时而停止。

with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:
    future_tasks = {executor.submit(self.crawl_task, url): url for url in self.requests.keys()}

    try:
        for future in concurrent.futures.as_completed(future_tasks, timeout=10):
            result = future.result()
            self.responses[result[0]] = result[1]
    except Exception as e:
        print(e)

timeout = [url for url in self.requests.keys() if url not in self.responses.keys()]

print('URL Threads timed out: ', timeout)

我必须指出，这违背了传统观念。通常，如果您将整个 for 循环包装在异常中，则循环中异常之后的任何内容都不应处理，但期货的魔力似乎允许循环中的所有内容（超时的线程除外）进行处理。

Answer 2

执行此操作的一种方法是在 self.crawl_task 中执行开始时将 url 记录到文件中。在线程任务完成之前，它可以附加一个字符串“DONE”，也许还有一个时间戳。

另外，您需要处理 TimeoutError 异常以不让执行中断。如果超时，您可以查看其中没有“DONE”字符串的文件日志。