Queue 示例 - 一个并发网络爬虫¶

Tornado 的 tornado.queues 模块对于协程实现了异步的 生产者 /消费者 模型, 实现了类似于 Python 标准库中线程中的 queue 模块.

一个协程 yield Queue.get 将会在队列中有值时暂停.如果队列设置了最大值, 协程会 yield Queue.put 暂停直到有空间来存放.

Queue 从零开始维护了一系列未完成的任务.put 增加计数; task_done 来减少它.

在这个网络爬虫的例子中, 队列开始仅包含 base_url. 当一个 worker 获取一个页面他会讲链接解析并将其添加到队列中,然后调用 task_done 来减少计数. 最后, 一个worker 获取到页面的 URLs 都是之前抓取过的, 队列中没有剩余的工作要做. worker调用 task_done 将计数减到0 . 主协程中等待 join, 取消暂停并完成.

  1. import time
  2. from datetime import timedelta
  3.  
  4. try:
  5.     from HTMLParser import HTMLParser
  6.     from urlparse import urljoin, urldefrag
  7. except ImportError:
  8.     from html.parser import HTMLParser
  9.     from urllib.parse import urljoin, urldefrag
  10.  
  11. from tornado import httpclient, gen, ioloop, queues
  12.  
  13. base_url = 'http://www.tornadoweb.org/en/stable/'
  14. concurrency = 10
  15.  
  16.  
  17. @gen.coroutine
  18. def get_links_from_url(url):
  19.     """Download the page at `url` and parse it for links.
  20.  
  21.     Returned links have had the fragment after `#` removed, and have been made
  22.     absolute so, e.g. the URL 'gen.html#tornado.gen.coroutine' becomes
  23.     'http://www.tornadoweb.org/en/stable/gen.html'.
  24.     """
  25.     try:
  26.         response = yield httpclient.AsyncHTTPClient().fetch(url)
  27.         print('fetched %s' % url)
  28.  
  29.         html = response.body if isinstance(response.body, str) \
  30.             else response.body.decode()
  31.         urls = [urljoin(url, remove_fragment(new_url))
  32.                 for new_url in get_links(html)]
  33.     except Exception as e:
  34.         print('Exception: %s%s' % (e, url))
  35.         raise gen.Return([])
  36.  
  37.     raise gen.Return(urls)
  38.  
  39.  
  40. def remove_fragment(url):
  41.     pure_url, frag = urldefrag(url)
  42.     return pure_url
  43.  
  44.  
  45. def get_links(html):
  46.     class URLSeeker(HTMLParser):
  47.         def __init__(self):
  48.             HTMLParser.__init__(self)
  49.             self.urls = []
  50.  
  51.         def handle_starttag(self, tag, attrs):
  52.             href = dict(attrs).get('href')
  53.             if href and tag == 'a':
  54.                 self.urls.append(href)
  55.  
  56.     url_seeker = URLSeeker()
  57.     url_seeker.feed(html)
  58.     return url_seeker.urls
  59.  
  60.  
  61. @gen.coroutine
  62. def main():
  63.     q = queues.Queue()
  64.     start = time.time()
  65.     fetching, fetched = set(), set()
  66.  
  67.     @gen.coroutine
  68.     def fetch_url():
  69.         current_url = yield q.get()
  70.         try:
  71.             if current_url in fetching:
  72.                 return
  73.  
  74.             print('fetching %s' % current_url)
  75.             fetching.add(current_url)
  76.             urls = yield get_links_from_url(current_url)
  77.             fetched.add(current_url)
  78.  
  79.             for new_url in urls:
  80.                 # Only follow links beneath the base URL
  81.                 if new_url.startswith(base_url):
  82.                     yield q.put(new_url)
  83.  
  84.         finally:
  85.             q.task_done()
  86.  
  87.     @gen.coroutine
  88.     def worker():
  89.         while True:
  90.             yield fetch_url()
  91.  
  92.     q.put(base_url)
  93.  
  94.     # Start workers, then wait for the work queue to be empty.
  95.     for _ in range(concurrency):
  96.         worker()
  97.     yield q.join(timeout=timedelta(seconds=300))
  98.     assert fetching == fetched
  99.     print('Done in %d seconds, fetched %s URLs.' % (
  100.         time.time() - start, len(fetched)))
  101.  
  102.  
  103. if __name__ == '__main__':
  104.     import logging
  105.     logging.basicConfig()
  106.     io_loop = ioloop.IOLoop.current()
  107.     io_loop.run_sync(main)

原文:

https://tornado-zh-cn.readthedocs.io/zh_CN/latest/guide/queues.html