如何在Python3中检测并发异常？

Question

由于并发的Futures模块，我刚刚转到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

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，则直到所有未完成的期货执行完毕并且与执行者关联的资源被释放后，该方法才会返回。如果wait为False，则此方法将立即返回，并且当所有未决的期货执行完毕时，与执行程序关联的资源将被释放。无论wait的值如何，整个Python程序都将在所有未完成的Future完成执行后才会退出。

   [如果使用with语句，可以避免必须显式调用此方法，这将关闭Executor（等待，就像在将Executor.shutdown()设置为True的情况下调用了C0]）

   就像在线程上调用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
...

Answer 2

参加聚会的方式为时已晚，但也许会对其他人有所帮助...

我很确定原来的问题没有得到真正的回答。人们感到不安的是，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()

执行上面的示例代码时，您将在屏幕上看到显示在打印语句中的文本，但是您将永远不会看到异常。这是因为每个线程的结果都保存在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()

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