python - 从 ThreadPoolExecutor 调用时对 urllib.request.urlopen 的异常调用会留下打开的文件描述符

Question

我正在尝试使用多个线程从雅虎财经下载大量数据。我正在使用 concurrent.futures.ThreadPoolExecutor 来加快速度。一切都很顺利，直到我消耗掉所有可用的文件描述符(默认为 1024)。

当 urllib.request.urlopen() 引发异常时，文件描述符保持打开状态(无论我使用的套接字超时时间是多少)。通常，如果我仅从单个(主)线程运行内容，则该文件描述符会被重用，因此不会发生此问题。但是，当从 ThreadPoolExecutor 线程进行这些异常的 urlopen() 调用时，这些文件描述符将保持打开状态。到目前为止，我提出的唯一解决方案是使用非常麻烦且低效的进程(ProcessPoolExecutor)，或者将允许的文件描述符的数量增加到非常大的数量(并非所有潜在用户无论如何，我的图书馆都会这样做)。必须有一种更聪明的方法来处理这个问题。

而且我想知道这是否是 Python 库中的错误，或者我只是做错了什么......

我在 Debian 上运行 Python 3.4.1(测试，内核 3.10-3-amd64)。

这是演示此行为的示例代码:

import concurrent
import concurrent.futures
import urllib.request
import os
import psutil
from time import sleep


def fetchfun(url):
    urllib.request.urlopen(url)


def main():

    print(os.getpid())
    p = psutil.Process(os.getpid())
    print(p.get_num_fds())


    # this url doesn't exist
    test_url = 'http://ichart.finance.yahoo.com/table.csv?s=YHOOxyz' + \
            '&a=00&b=01&c=1900&d=11&e=31&f=2019&g=d'

    with concurrent.futures.ThreadPoolExecutor(1) as executor:
        futures = []
        for i in range(100):
            futures.append(executor.submit(fetchfun, test_url))
        count = 0
        for future in concurrent.futures.as_completed(futures):
            count += 1
            print("{}: {} (ex: {})".format(count, p.get_num_fds(), future.exception()))

    print(os.getpid())
    sleep(60)


if __name__ == "__main__":
    main()

Answer 1

当 HTTPError被引发，它保存对 HTTPResponse 的引用请求的对象为 fp HTTPError 的属性。该引用将保存在您的 futures 中列表，在程序结束之前该列表不会被销毁。这意味着有对 HTTPResponse 的引用。在整个程序中保持活力。只要该引用存在， HTTPResponse 中使用的套接字保持开放。解决此问题的一种方法是显式关闭 HTTPResponse当你处理异常时:

with concurrent.futures.ThreadPoolExecutor(1) as executor:
    futures = []
    for i in range(100):
        futures.append(executor.submit(fetchfun, test_url))
    count = 0
    for future in concurrent.futures.as_completed(futures):
        count += 1
        exc = future.exception()
        print("{}: {} (ex: {})".format(count, p.get_num_fds(), exc))
        exc.fp.close()  # Close the HTTPResponse