ConCurent.futures.ThreadPoolExecutor/多线程内存不足(已终止)

问题描述

我目前正在做一个据说很容易的Web抓取项目，同时学习Python。我有一个大约70MB的列表，其中有几百万个IP地址(sys.argv[1])需要处理。当然，并不是所有这些服务都是可访问的。

我正在尝试使用并发。未来，但当前遇到内存问题-最终导致整个进程被终止。

现在，我已经按照here的建议将我的期货分成了两组(完成和未完成)。 我正在使用大约100个工作进程(sys.argv[2])，并且有1 GB的可用内存。

我以为所有完成的期货都会在Future.Result()被调用时被释放，并且=&>期货1000完成了？然而，它似乎只是减慢了进程(包括内存被填满，直到进程被终止)。

我在这里错过了什么？对如何处理这一问题有什么建议吗？

提前谢谢您。

我的代码如下：

import sys
import requests
import concurrent.futures
import urllib3
from lxml.html import fromstring
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def title(host):
    try:
        url="https://"+host
        r = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=3, verify=False)
        tree = fromstring(r.content.decode('utf-8'))
        title = tree.findtext('.//title')
        print(host+": "+title)
    except:
        pass

max=int(sys.argv[2])
with concurrent.futures.ThreadPoolExecutor(max_workers=max) as executor:
    futures_done = set()
    futures_notdone = set()
    with open(sys.argv[1]) as f:
        for line in f:
            host = line.strip()
            futures_notdone.add(executor.submit(title, host))
            if len(futures_notdone) >= max:
                done, futures_notdone = concurrent.futures.wait(futures_notdone, return_when=concurrent.futures.FIRST_COMPLETED)
                futures_done.update(done)
            for future in futures_done:
                if len(futures_done) >= 1000:
                    future.result()

---

解决方案

看起来像是将完成的期货存储在一个集合中，而不是在以后清除此列表，因此它可能会变得非常大。这可能是您的记忆问题的原因。未来的.release()方法不释放它，它仍然在done_future列表中被引用。

不完美，但您可以尝试以下方法。它最多调度max个作业并发执行。它定期收集已完成的工作，并重新安排新的工作。idea来自本博客。

我在此方法中看到的缺点是，它必须定期轮询max计划的作业以查找已完成的作业，如果max值较大，这可能会很慢。

import sys
import requests
import concurrent.futures
import urllib3
from itertools import islice
from lxml.html import fromstring
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def title(host: str) -> str:
    try:
        url="https://"+host
        r = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=3, verify=False)
        tree = fromstring(r.content.decode('utf-8'))
        title = tree.findtext('.//title')
        return host+": "+title
    except:
        pass

max = int(sys.argv[2])

with concurrent.futures.ThreadPoolExecutor(max_workers=max) as executor:
    with open(sys.argv[1]) as f:
        futures = {executor.submit(title, h) for h in islice(f, max)}
        
        while futures:
            done, futures = concurrent.futures.wait(
                futures, return_when=concurrent.futures.FIRST_COMPLETED)

            for future in done:
                print(future.result())

            for h in islice(f, len(done)):
                futures.add(executor.submit(title, h))

---

这是一个可能对您有用的解决方法，它在我的计算机上运行了100多万次迭代，而没有使用超过150个MO。

它只是一个带有两个队列的自定义线程池，用于管理并发资源访问和限制最大并发。

import sys
from typing import Optional
import requests
import urllib3
from lxml.html import fromstring
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
from queue import Queue
from threading import Thread


def get_title(host: str) -> Optional[str]:
    try:
        url = f"https://{host}"
        r = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=1, verify=False)
        tree = fromstring(r.content.decode('utf-8'))
        title = tree.findtext('.//title')
        return f"{host}: {title}"
    except Exception:
        return None

class Pool:
    def __init__(self, work, max_concurrent_jobs, max_worker: int = 32) -> None:
        self.max_workers = max_worker
        self.work_queue = Queue(max_concurrent_jobs)
        self.out_queue = Queue()
        self.is_running = True

        def _work():
            while self.is_running:
                item = self.work_queue.get()
                result = work(item)
                self.work_queue.task_done()
                self.out_queue.put(result)

        for _ in range(max_worker):
            Thread(target=_work).start()

    def close(self):
        self.is_running = False


if __name__ == "__main__":
    file_name = sys.argv[1]
    max = int(sys.argv[2])
    pool = Pool(work=get_title, max_concurrent_jobs=max)

    def worker():
        while True:
            item = pool.out_queue.get()
            if item is not None:
                print(item) # Or any follow-up job
            pool.out_queue.task_done()

    Thread(target=worker, daemon=True).start()

    with open(file_name) as f:
        for h in f:
            pool.work_queue.put(h.strip())