龙卷风：来自迭代器的 AsyncHttpClient.fetch？答案

【问题标题】：tornado: AsyncHttpClient.fetch from an iterator?龙卷风：来自迭代器的 AsyncHttpClient.fetch？
【发布时间】：2015-08-05 15:20:44
【问题描述】：

我正在尝试编写网络爬虫，并希望尽快发出 HTTP 请求。 tornado's AsyncHttpClient 似乎是个不错的选择，但我见过的所有示例代码（例如 https://stackoverflow.com/a/25549675/1650177）基本上都在大量 URL 上调用 AsyncHttpClient.fetch，让 tornado 将它们排队并最终发出请求。

但是，如果我想处理来自文件或网络的无限长（或只是非常大）的 URL 列表怎么办？我不想将所有 URL 加载到内存中。

谷歌搜索，但似乎无法从迭代器中找到AsyncHttpClient.fetch 的方法。然而，我确实找到了一种使用 gevent 做我想做的事情的方法：http://gevent.org/gevent.threadpool.html#gevent.threadpool.ThreadPool.imap。有没有办法在龙卷风中做类似的事情？

我想到的一个解决方案是最初只排队这么多 URL，然后添加逻辑以在 fetch 操作完成时排队更多，但我希望有一个更清洁的解决方案。

任何帮助或建议将不胜感激！

【问题讨论】：

您链接到的示例问题不会将整个 url 列表加载到内存中 - 它只是一次从文件中读取一行。您是否只是担心一次打开太多 http 连接？如果是这样，我不确定在生成器函数内部进行调用会对您有什么帮助。您能否准确说明您在寻找什么？
我链接到的示例问题在urls.txt 中的每个URL 上一次调用fetch，是的，但这是在内部为列表中的每个URL 排队一个HTTP 请求。我不担心打开太多 HTTP 连接，而是担心有太多 HTTP 请求排队。
在 gevent 中找到我想用龙卷风做的事情，如果它有助于澄清事情：gevent.org/…。不过，我还是想知道你是如何在龙卷风中做类似的事情的！
AsyncHttpClient 是异步的 - 请求没有排队，它们都使用非阻塞 I/O 并行执行。当请求完成时，handle_request 回调被执行。唯一会同步发生的事情是在检索到响应后执行回调。
也许我在这里遗漏了一些东西，但我很确定他们是。直接来自文档 (tornado.readthedocs.org/en/latest/…)：max_clients is the number of concurrent requests that can be in progress; when this limit is reached additional requests will be queued. max_clients 默认设置为 10，我不想将其设置为 len(list(iter))。

标签： python asynchronous tornado

【解决方案1】：

我会在 https://github.com/tornadoweb/tornado/blob/master/demos/webspider/webspider.py 的一个变体中使用一个队列和多个工作人员来完成此操作

import tornado.queues
from tornado import gen
from tornado.httpclient import AsyncHTTPClient
from tornado.ioloop import IOLoop

NUM_WORKERS = 10
QUEUE_SIZE = 100
q = tornado.queues.Queue(QUEUE_SIZE)
AsyncHTTPClient.configure(None, max_clients=NUM_WORKERS)
http_client = AsyncHTTPClient()

@gen.coroutine
def worker():
    while True:
        url = yield q.get()
        try:
            response = yield http_client.fetch(url)
            print('got response from', url)
        except Exception:
            print('failed to fetch', url)
        finally:
            q.task_done()

@gen.coroutine
def main():
    for i in range(NUM_WORKERS):
        IOLoop.current().spawn_callback(worker)
    with open("urls.txt") as f:
        for line in f:
            url = line.strip()
            # When the queue fills up, stop here to wait instead
            # of reading more from the file.
            yield q.put(url)
    yield q.join()

if __name__ == '__main__':
    IOLoop.current().run_sync(main)

【讨论】：

正是我想要的谢谢！应该检查演示。