如何在 Python3 中检测 concurrent.futures 中的异常？

Question

由于它的并发期货模块，我刚刚转向 python3。我想知道是否可以让它检测错误。我想使用并发期货来并行程序，如果有更高效的模块请告诉我。

我不喜欢多处理，因为它太复杂而且没有多少文档可用。但是，如果有人可以编写一个没有类的 Hello World，只使用多处理并行计算的函数，这样它就很容易理解了，那就太好了。

这是一个简单的脚本：

from concurrent.futures import ThreadPoolExecutor

def pri():
    print("Hello World!!!")

def start():
    try:
        while True:
            pri()
    except KeyBoardInterrupt:
        print("YOU PRESSED CTRL+C")


with ThreadPoolExecutor(max_workers=3) as exe:
    exe.submit(start)

以上代码只是一个演示，说明 CTRL+C 如何无法打印语句。

我想要的是能够调用函数是存在错误。这种错误检测必须来自函数本身。

另一个例子

import socket
from concurrent.futures import ThreadPoolExecutor 
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
def con():
    try:
        s.connect((x,y))
        main()
    except: socket.gaierror
         err()
def err():
    time.sleep(1)
    con()
def main():
    s.send("[+] Hello")
with ThreadPoolExecutor as exe:
    exe.submit(con)

Answer 1

聚会太晚了，但也许它会帮助别人......

我很确定最初的问题没有得到真正的回答。人们对 user5327424 正在使用键盘中断引发异常这一事实感到困惑，而关键是没有引发异常（无论它是由什么引起的）。例如：

import concurrent.futures


def main():
    numbers = range(10)

    with concurrent.futures.ThreadPoolExecutor() as executor:
        results = {executor.submit(raise_my_exception, number): number for number in numbers}


def raise_my_exception(number):
    print('Proof that this function is getting called. %s' % number)
    raise Exception('This never sees the light of day...')


main()

当执行上面的示例代码时，您会看到屏幕上显示 print 语句中的文本，但您永远不会看到异常。这是因为每个线程的结果都保存在results 对象中。您需要迭代该对象以获取您的异常。以下示例显示了如何访问结果。

import concurrent.futures


def main():
    numbers = range(10)

    with concurrent.futures.ThreadPoolExecutor() as executor:
        results = {executor.submit(raise_my_exception, number): number for number in numbers}

    for result in results:
        # This will cause the exception to be raised (but only the first one)
        print(result.result())


def raise_my_exception(number):
    print('Proof that this function is getting called. %s' % number)
    raise Exception('This will be raised once the results are iterated.')


main()

我不确定我是否喜欢这种行为，但它确实允许线程完全执行，而不管各个线程中遇到的异常。

Answer 2

Here 是一个解决方案。我不确定你是否喜欢它，但我想不出其他的。我已经修改了您的代码以使其正常工作。

from concurrent.futures import ThreadPoolExecutor
import time

quit = False

def pri():
    print("Hello World!!!")

def start():
    while quit is not True:
        time.sleep(1)
        pri()

try:
    pool = ThreadPoolExecutor(max_workers=3)
    pool.submit(start)

    while quit is not True:
        print("hei")
        time.sleep(1)
except KeyboardInterrupt:
    quit = True

以下是要点：

1. 当您使用with ThreadPoolExecutor(max_workers=3) as exe 时，它会等待所有任务完成。看看Doc

   如果wait 为True，则此方法将不会返回，直到所有未决的期货都执行完毕并且与执行程序相关的资源已被释放。如果等待是False，则此方法将立即返回，并且当所有未决的期货执行完毕后，与执行器关联的资源将被释放。无论 wait 的值如何，整个 Python 程序都不会退出，直到所有未决的期货都执行完毕。

   如果您使用with 语句，您可以避免显式调用此方法，这将关闭Executor（等待就像调用Executor.shutdown() 并设置为True）

   这就像在线程上调用join()。
   这就是为什么我将其替换为：

   pool = ThreadPoolExecutor(max_workers=3)
   pool.submit(start)
   

2. 主线程必须在做“工作”才能捕捉到 Ctrl+C。所以你不能把主线程放在那里然后退出，最简单的方法是运行一个无限循环

3. 现在你在主线程中运行了一个循环，当你点击CTRL+C时，程序将进入except KeyboardInterrupt块并设置quit=True。然后你的工作线程就可以退出了。

严格来说，这只是一种解决方法。在我看来，这不可能有其他方法。

编辑
我不确定是什么在困扰您，但您可以毫无问题地在另一个线程中捕获异常：

import socket
import time
from concurrent.futures import ThreadPoolExecutor 
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

def con():
    try:
        raise socket.gaierror
        main()
    except socket.gaierror:
        print("gaierror occurred")
        err()

def err():
    print("err invoked")
    time.sleep(1)
    con()

def main():
    s.send("[+] Hello")

with ThreadPoolExecutor(3) as exe:
    exe.submit(con)

输出

gaierror occurred
err invoked
gaierror occurred
err invoked
gaierror occurred
err invoked
gaierror occurred
...