TimeoutError給定以下在某些時候導致 a 的簡單方法。如何確定導致超時的條件（例如引數或其他內容）是什么？
In [82]: def f(i):
...: print(i)
...: if i == 2: time.sleep(1)
...: return "done"
...:
In [83]: with ThreadPoolExecutor(max_workers=10) as ex:
...: res = ex.map(f, range(10), timeout=0.001)
...: try:
...: for r in res:
...: print(f"{r=}")
...: except Exception as e:
...: print(f"{e=}")
...:
0
1
2
3
4
5
6
8
7
r='done'
9
r='done'
e=TimeoutError()
TimeoutErrorin 的例外處理f不起作用，因為TimeoutError在遍歷結果集時引發了。
In [84]: with ThreadPoolExecutor(max_workers=10) as ex:
...: res = ex.map(f, range(10), timeout=0.001)
...: for r in res:
...: print(f"{r=}")
...:
0
1
2
3
4
5
6
7
r='done'
r='done'
8
9
---------------------------------------------------------------------------
TimeoutError Traceback (most recent call last)
Cell In [84], line 3
1 with ThreadPoolExecutor(max_workers=10) as ex:
2 res = ex.map(f, range(10), timeout=0.001)
----> 3 for r in res:
4 print(f"{r=}")
File /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/concurrent/futures/_base.py:621, in Executor.map.<locals>.result_iterator()
619 yield fs.pop().result()
620 else:
--> 621 yield fs.pop().result(end_time - time.monotonic())
622 finally:
623 for future in fs:
File /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/concurrent/futures/_base.py:446, in Future.result(self, timeout)
444 return self.__get_result()
445 else:
--> 446 raise TimeoutError()
447 finally:
448 # Break a reference cycle with the exception in self._exception
449 self = None
TimeoutError:
uj5u.com熱心網友回復：
例外本身不攜帶任何相關資訊。但是，正如檔案所述：
回傳的迭代器引發一個
concurrent.futures.TimeoutErrorif__next__()被呼叫，并且在從原始呼叫超時秒后結果不可用Executor.map()
因此，例外始終與您嘗試檢索的下一個元素相關聯。如果您跟蹤這些論點，您就可以知道是哪一個導致了問題。
import time
from concurrent import futures
def f(i):
time.sleep(i)
return i
def run_checked(fun, args):
args = list(args)
idx = -1
with futures.ThreadPoolExecutor(max_workers=4) as tp:
try:
for idx, result in enumerate(tp.map(fun, args, timeout=4)):
print(result)
except futures.TimeoutError:
print("Timeout on arguments '%s'" % args[idx 1])
if __name__ == '__main__':
run_checked(f, range(10))
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/529625.html