Python 中的多处理 HTTP 获取请求

Question

我必须向大量网站发出大量（数千）HTTP GET 请求。这很慢，因为某些网站可能没有响应（或需要很长时间才能响应），而其他网站则超时。因为我需要尽可能多的回复，所以设置一个小的超时（3-5 秒）对我不利。

我还没有在 Python 中进行任何类型的多处理或多线程处理，而且我已经阅读文档有一段时间了。这是我到目前为止所拥有的：

import requests
from bs4 import BeautifulSoup
from multiprocessing import Process, Pool

errors = 0

def get_site_content(site):
    try :
        # start = time.time()
        response = requests.get(site, allow_redirects = True, timeout=5)
        response.raise_for_status()
        content = response.text
    except Exception as e:
        global errors
        errors += 1
        return ''
    soup = BeautifulSoup(content)
    for script in soup(["script", "style"]):
        script.extract()
    text = soup.get_text()

    return text

sites = ["http://www.example.net", ...]

pool = Pool(processes=5)
results = pool.map(get_site_content, sites)
print results

现在，我希望以某种方式加入返回的结果。这允许两种变化：

1. 每个进程都有一个本地列表/队列，其中包含它积累的内容，并与其他队列连接以形成单个结果，其中包含所有站点的所有内容。

2. 每个进程在执行过程中都会写入一个全局队列。这将需要一些用于并发检查的锁定机制。

多处理或多线程在这里是更好的选择吗？我将如何使用 Python 中的任何一种方法完成上述任务？

---

编辑：

我确实尝试过类似以下的事情：

# global
queue = []
with Pool(processes = 5) as pool:
    queue.append(pool.map(get_site_contents, sites))

print queue

但是，这给了我以下错误：

with Pool(processes = 4) as pool:
AttributeError: __exit__

我不太明白。我有一个小麻烦了解什么确切pool.map确实，过去在迭代的第二个参数应用功能中的每个对象上。它会返回任何东西吗？如果没有，我是否从函数内部附加到全局队列？

Answer 1

我在大学里有过类似的作业（实现一个多进程网络爬虫），并使用了 python 多处理库中的多处理安全 Queue 类，它将通过锁和并发检查发挥所有作用。文档中的示例指出：

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    mp.set_start_method('spawn')
    q = mp.Queue()
    p = mp.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

但是，我必须编写一个单独的进程类才能使其工作，因为我希望它工作。而且我没有使用进程池。相反，我尝试检查内存使用情况并生成一个进程，直到达到预设的内存阈值。

Answer 2

pool.map启动 'n' 个进程，这些进程接受一个函数并使用 iterable 中的一个项目运行它。当这样的过程完成并返回时，返回值存储在结果列表中与输入变量中的输入项相同的位置。

例如：如果编写一个函数来计算数字的平方，然后pool.map使用 a 对数字列表运行此函数。def square_this(x): square = x**2 return square

input_iterable = [2, 3, 4]
pool = Pool(processes=2) # Initalize a pool of 2 processes
result = pool.map(square_this, input_iterable) # Use the pool to run the function on the items in the iterable
pool.close() # this means that no more tasks will be added to the pool
pool.join() # this blocks the program till function is run on all the items
# print the result
print result

...>>[4, 9, 16]

该Pool.map技术在您的情况下可能并不理想，因为它会阻塞直到所有进程完成。即，如果网站没有响应或响应时间过长，您的程序将卡在等待它。而是尝试multiprocessing.Process在您自己的类中对进行子类化，该类会轮询这些网站并使用队列来访问结果。当您有满意数量的响应时，您可以停止所有进程，而无需等待剩余的请求完成。