requests 线程池无法中止子线程问题

from concurrent.futures import ThreadPoolExecutor, as_completed import requests from tqdm import tqdm def download(url): try: filesize = int(requests.get(url, stream=True).headers['Content-length']) or 0 filename = requests.get(url, stream=True).headers['content-disposition'] with requests.get(url, stream=True) as responese: with tqdm( total=filesize, desc=f"Downloading: ", unit="B", unit_divisor=1024, ascii=True, unit_scale=True, ) as bar: with open(filename, mode="ab") as f: # 追加模式 for chunk in responese.iter_content(1024): f.write(chunk) bar.update(len(chunk)) except (SystemExit, KeyboardInterrupt): raise "KeyboardInterrupt" urls = [...] with ThreadPoolExecutor(maxworkers=8) as pool: tasks = pool.map(download, urls) for task in as_completed(tasks): print(task.result())

Supplement 1 Aug 8, 2023

更准确的提问应该是：concurrent.futures 的 submit、map 方法提交后，无法捕获 Ctrl + C 信号终止子线程简单测试 delay server 如下：

import time from fastapi import FastAPI from fastapi.responses import FileResponse from uvicorn import run app = FastAPI() @app.get("/") def hello(): return {"data": "hello world"} @app.get("/pic/{file}") def d(file: str): time.sleep(60) return FileResponse(f"pic/{file}") if __name__ == "__main__": run("download_server:app", host="localhost", port=8000, reload=True)

6 replies 2023-08-11 10:54:01 +08:00

t133

Aug 8, 2023 via iPhone

指定 daemon thread 可解

evemoo

Aug 8, 2023

@t133
实际测试，join() 还要带 timeout 参数才能 raise "KeyboardInterrupt"

```python
tasks = []
for url in urls:
task = threading.Thread(target=download, args=(url,))
task.setDaemon(True)
tasks.append(task)
task.start()

for task in tasks:
task.join(2)
```

evemoo

Aug 8, 2023

搜了下 concurrent.futrues 的 daemon ，没有找到相似解法。

ruanimal

Aug 8, 2023

你需要的是 graceful shutdown 吧，看看 signal 处理

ClericPy

Aug 8, 2023

和 Requests 无关, 多线程本来就很难终止, 不然我也不会 all-in 协程去了.

chaoshui

Aug 11, 2023

如果不关心子线程的任务情况的话，可以直接在主线程关闭线程池 pool. shutdown(wait=False) 。
如果关心子线程的任务情况，比如正在下载的文件继续下载完毕或者怎样，可以通过一个 threading.Event 通知正在进行的任务退出，配合 Future 对象的 cancel 方法可以干净地处理所有的任务。