时间戳

首页 > 解决方案 > 中断线程池执行器

python - 中断线程池执行器

问题描述

当脚本被中断时，如何立即停止concurrent.futures.ThreadPoolExecutor并丢弃任何挂起的操作，如下例所示：

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed


def get_parallel(arguments_list: list[dict]):
    try:
        with ThreadPoolExecutor(max_workers=25) as executor:
            futures_buffer = [executor.submit(requests.get, **kwargs) for kwargs in arguments_list]

            for future in as_completed(futures_buffer):
                try:
                    response = future.result()
                    print (response.url)
                    yield response.url, response.status_code, response.json()['args']
                except KeyboardInterrupt:
                    executor.shutdown(wait=False, cancel_futures=True)
                    yield 'KeyboardInterrupt 1'
                    return
                except Exception as exception:
                    yield exception
    except KeyboardInterrupt:
        yield 'KeyboardInterrupt 2'
        return


if __name__ == '__main__':
    arguments = [dict(url=f'https://httpbin.org/get?q={i}') for i in range(200)]
    for t in get_parallel(arguments):
        print(t)

就像现在的代码一样，当我从终端运行它然后调用 ^C 时，它会停止打印结果，但会挂起，就像它没有被中断一样，最后它会打印KeyboardInterrupt 2。

标签： pythonconcurrent.futures

解决方案

您不能在 python 中向单个线程发出信号，即使可以，也不能保证请求没有创建自己的线程作为工作线程。您可以终止进程，因此您可以将请求委托给子进程并在遇到键盘中断时终止它。最简洁的方法是管理您自己的子流程和工作项队列。但是，如果您不介意一点黑客攻击，请concurrent.futures.ProcessPoolExecutor保留其池进程的列表，您可以劫持它。但它的黑客行为......它可能会在未来的某个时间打破。由于响应被腌制并发送回父进程，最好有一个从响应对象而不是响应对象本身获取有用数据的中间工作函数。

import concurrent.futures
import os
import requests
import signal

def worker(arguments_list: dict):
    """use requests to get web page and return url, status, json args"""
    resp = requests.get(**arguments_list)
    # todo: when status_code not 200 and json decode fails, do...???
    return resp.url, resp.status_code, resp.json()['args']

def get_parallel(arguments_list: list[dict]):
    with concurrent.futures.ProcessPoolExecutor(max_workers=25) as executor:
        try:
            futures_buffer = [executor.submit(worker, kwargs) for kwargs in arguments_list]

            for future in concurrent.futures.as_completed(futures_buffer):
                    url, status_code, args = future.result()
                    print (url)
                    yield url, status_code, args
        except KeyboardInterrupt:
            for pid in executor._processes:
                os.kill(pid, signal.SIGKILL)
            yield 'KeyboardInterrupt 2'

if __name__ == '__main__':
    arguments = [dict(url=f'https://httpbin.org/get?q={i}') for i in range(200)]
    for t in get_parallel(arguments):
        print(t)

推荐阅读