Python ThreadPoolExecutor 终止所有线程

Question

我正在运行一段 python 代码，其中多个线程通过线程池执行程序运行。每个线程都应该执行一项任务（例如获取网页）。我想要做的是终止所有线程，即使其中一个线程失败。例如：

with ThreadPoolExecutor(self._num_threads) as executor:
    jobs = []
    for path in paths:
        kw = {"path": path}
        jobs.append(executor.submit(start,**kw))
    for job in futures.as_completed(jobs):
        result = job.result()
        print(result)
def start(*args,**kwargs):
    #fetch the page
    if(success):
        return True
    else:
        #Signal all threads to stop

有可能吗？除非所有线程都成功，否则线程返回的结果对我来说毫无用处，因此即使其中一个失败，我想节省其余线程的一些执行时间并立即终止它们。实际的代码显然是在做相对冗长的任务，有几个失败点。

Answer 1

如果您已经完成了线程并想研究进程，那么这里的代码和平看起来非常有前途和简单，几乎与线程相同的语法，但使用多处理模块。

当超时标志到期时进程终止，非常方便。

import multiprocessing

def get_page(*args, **kwargs):
    # your web page downloading code goes here

def start_get_page(timeout, *args, **kwargs):
    p = multiprocessing.Process(target=get_page, args=args, kwargs=kwargs)
    p.start()
    p.join(timeout)
    if p.is_alive():
        # stop the downloading 'thread'
        p.terminate()
        # and then do any post-error processing here

if __name__ == "__main__":
    start_get_page(timeout, *args, **kwargs)

Answer 2

您可以尝试使用func-timeout 中的StoppableThread。 但是终止线程是强烈 discouraged。如果你需要杀死一个线程，你可能有一个设计问题。查看替代方案：asyncio 协程和 multiprocessing 具有合法的取消/终止功能。

Answer 3

我已经为我遇到的类似问题创建了一个答案，我认为该答案适用于这个问题。

• Terminate executor using ThreadPoolExecutor from concurrent.futures module

from concurrent.futures import ThreadPoolExecutor, as_completed
from time import sleep

NUM_REQUESTS = 100


def long_request(id):
    sleep(1)

    # Simulate bad response
    if id == 10:
        return {"data": {"valid": False}}
    else:
        return {"data": {"valid": True}}


def check_results(results):
    valid = True
    for result in results:
        valid = result["data"]["valid"]

    return valid


def main():
    futures = []
    responses = []
    num_requests = 0

    with ThreadPoolExecutor(max_workers=10) as executor:
        for request_index in range(NUM_REQUESTS):
            future = executor.submit(long_request, request_index)

            # Future list
            futures.append(future)

        for future in as_completed(futures):

            is_responses_valid = check_results(responses)

            # Cancel all future requests if one invalid
            if not is_responses_valid:
                executor.shutdown(wait=False)
            else:
                # Append valid responses
                num_requests += 1
                responses.append(future.result())

    return num_requests


if __name__ == "__main__":
    requests = main()
    print("Num Requests: ", requests)

Answer 4

我会这样做：

import concurrent.futures

def start(*args,**kwargs):
    #fetch the page
    if(success):
        return True
    else:
        return False

with concurrent.futures.ProcessPoolExecutor() as executor:
    results = [executor.submit(start, {"path": path}) for path in paths]
    concurrent.futures.wait(results, timeout=10, return_when=concurrent.futures.FIRST_COMPLETED)
    for f in concurrent.futures.as_completed(results):
        f_success = f.result()
        if not f_success:
            executor.shutdown(wait=False, cancel_futures=True) # shutdown if one fails
        else:
            #do stuff here

如果任何结果不为真，一切都会立即关闭。

Answer 5

在我的代码中我使用了多处理

import multiprocessing as mp
pool = mp.Pool()
for i in range(threadNumber):
    pool.apply_async(publishMessage, args=(map_metrics, connection_parameters...,))

pool.close()
pool.terminate()