python futures.as_completed在发生错误后停止响应

Question

我有一些使用Future的代码可以正常工作并提供输出，直到遇到反复错误然后输出停止为止。其余过程将继续执行（我可以看到它们正在数据库中运行），我只是无法捕获结果。我想念什么？

from concurrent import futures
import oracle_manager
import os

sql_statements=['select count(*) from PS_SCC_NTF_PREF',
'select count(*) from PS_SCC_STN_LTR_TBL',
'select count(*) from PS_SCC_TM_TRAN_AGR',
'select count(*) from PS_SCRTY_TBL_INST',
'select count(*) from PS_SCTN_CMBND',
'select count(*) from PS_SCTN_CMBND_TBL',
'select count(*) from PS_SEC_ITEM_CLS',
'select count(*) from PS_SESSION_TBL',
...]

def run_sql(sql):
  temp_connection = oracle_manager.OracleConnection(un, pw, tns)
  value = temp_connection.execute_count(sql)
  return (value, os.getpid())

worker_pool = futures.ProcessPoolExecutor(5)
wait_for = [worker_pool.submit(run_sql, sql_statement) for sql_statement in sql_statements] 

for f in futures.as_completed(wait_for):
  try:
    print(f.result())
  except Exception as e:
    print(e)

((1069,), 23010)
((130464,), 23013)
((40220,), 23012)
((296372,), 23009)
((930954,), 23011)
((13836,), 23012)
((19707,), 23009)
((52130,), 23010)
((130707,), 23013)
((21,), 23011)
((276,), 23009)
((53157,), 23012)
((111,), 23009)
((381,), 23009)
An exception occurred running the following sql statement: "select count(*) from PS_SF_ACCTG_LN"
Oracle-Error-Code/Message: 600  ORA-00600: internal error code, arguments: [kskqgetserviceid:nosvcidpdb1], [3], [0], [], [], [], [], [], [], [], [], []

感谢您的帮助！

Answer 1

您可以在函数中使用catch the error，也可以在检索结果时捕获它。

使用玩具示例：

import concurrent.futures
import sys

def f(n):
    try:
        if n == 6:
            raise ValueError
    except ValueError as e:
        n = f'{n} raised a ValueError'
    return n

或

def f(n):
    if n == 6:
        raise ValueError
    return n

if __name__ == '__main__':
    with concurrent.futures.ProcessPoolExecutor() as executor:
        fs = [executor.submit(f,n) for n in range(10)]
    for f in concurrent.futures.as_completed(fs):
        try:
            print(f.result())
        except ValueError as e:
            print(f'{f}', file = sys.stderr)