Python - 多个同时发生的线程池

Question

我正在使用httplib2和lxml在python中编写一个web scraper(是的 - 我知道我可以使用scrapy.让我们移过那个......)刮刀有大约15000个页面可以解析为大约400,000个项目.我有代码解析项目即时运行(差不多)但从服务器下载页面的部分仍然非常慢.我想通过并发来克服这个问题.但是,我不能依赖每个页面都需要解析.我尝试过使用单个ThreadPool(比如multiprocessing.pool,但是完成了线程 - 这应该没问题,因为这是一个I/O绑定进程),但我想不出一种优雅(或工作)的方式当最后一个索引项的日期大于我们正在处理的项时停止的所有线程.现在,我正在研究一个使用ThreadPool的两个实例的方法 - 一个用于下载每个页面,另一个用于解析页面.简化的代码示例是:

#! /usr/bin/env python2

import httplib2
from Queue import PriorityQueue
from multiprocessing.pool import ThreadPool
from lxml.html import fromstring

pages = [x for x in range(1000)]
page_queue = PriorityQueue(1000)

url = "http://www.google.com"

def get_page(page):
    #Grabs google.com
    h = httplib2.Http(".cache")
    resp, content = h.request(url, "GET")
    tree = fromstring(str(content), base_url=url)
    page_queue.put((page, tree))
    print page_queue.qsize()

def parse_page():
    page_num, page = page_queue.get()
    print "Parsing page #" + str(page_num)
    #do more stuff with the page here
    page_queue.task_done()

if __name__ == "__main__":
    collect_pool = ThreadPool()
    collect_pool.map_async(get_page, pages)
    collect_pool.close()

    parse_pool = ThreadPool()
    parse_pool.apply_async(parse_page)
    parse_pool.close()


     parse_pool.join()
     collect_pool.join()
     page_queue.join()

然而,运行此代码并不能达到我的预期 - 即触发两个线程池:一个填充队列,另一个从中拉出来解析.它开始收集池并运行它然后开始parse_pool并运行它(我假设,我没有让代码运行足够长的时间来到parse_pool - 重点是collect_pool似乎正在运行).我很确定我已经搞砸了加入()的调用顺序,但我不能为我的生活弄清楚它们应该是什么顺序.我的问题基本上是这样的:我在这里咆哮着正确的树吗？如果是的话,我到底做错了什么？如果我不是 - 你的建议是什么？

Answer 1

首先,您的设计似乎在高层次上是正确的.使用线程池来收集页面是由httlib2模块的同步特性所证明的.(使用异步库,一个线程就足够了;请注意,即使使用httplib2,并且池中最多只有一个收集器线程因GIL而在任何时间运行.)解析池由lxml模块用C /写入来证明是合理的C++(并假设在解析页面期间释放全局解释器锁 - 这将在lxml文档或代码中进行检查!).如果后者不成立,那么通过拥有专用的解析池将不会获得性能,因为只有一个线程能够获取GIL.在这种情况下,最好使用进程池.

我不熟悉ThreadPool实现,但我认为它类似于多处理模块中的Pool类.在此基础上,问题似乎是您只为parse_pool创建一个工作项,并且在parse_page处理完第一个页面后,它从不尝试从那里出列其他页面.其他工作项也未提交到此池,因此处理停止,并且在parse_pool.close()调用之后(空)池的线程终止.

解决方案是消除page_queue.get_page()函数应该在parse_pool上放置一个工作项,方法是为它收集的每个页面调用apply_async(),而不是将它们提供给page_queue.

主线程应该等到collect_queue为空(即返回collect_pool.join()调用),然后它应该关闭parse_pool(因为我们可以确定不再为解析器提交任何工作).然后它应该等待parse_pool变为空,通过调用parse_pool.join()然后退出.

此外,您需要增加connect_pool中的线程数,以便同时处理更多的http请求.池中的默认线程数是CPU数量; 目前你不能发出超过那么多的请求.您可以尝试高达数千或数千的值; 观察池的CPU消耗; 它不应该接近1 CPU.