龙卷风 AsyncHTTPClient.fetch 异常

Question

我正在使用 tornado.httpclient.AsyncHTTPClient.fetch 从列表中获取域。当我以一些大的间隔（例如 500）来获取域时，一切都很好，但是当我将间隔减少到 100 时，下一个异常会时不时地发生：

Traceback (most recent call last):
  File "/home/crchemist/python-2.7.2/lib/python2.7/site-packages/tornado/simple_httpclient.py", line 289, in cleanup
    yield
  File "/home/crchemist/python-2.7.2/lib/python2.7/site-packages/tornado/stack_context.py", line 183, in wrapped
    callback(*args, **kwargs)
  File "/home/crchemist/python-2.7.2/lib/python2.7/site-packages/tornado/simple_httpclient.py", line 384, in _on_chunk_length
    self._on_chunk_data)
  File "/home/crchemist/python-2.7.2/lib/python2.7/site-packages/tornado/iostream.py", line 180, in read_bytes
    self._check_closed()
  File "/home/crchemist/python-2.7.2/lib/python2.7/site-packages/tornado/iostream.py", line 504, in _check_closed
    raise IOError("Stream is closed")
IOError: Stream is closed

这种行为的原因是什么？代码如下所示：

def fetch_domain(domain):
    http_client = AsyncHTTPClient()
    request = HTTPRequest('http://' + domain,
       user_agent=CRAWLER_USER_AGENT)
    http_client.fetch(request, handle_domain)


class DomainFetcher(object):
    def __init__(self, domains_iterator):
        self.domains = domains_iterator

    def __call__(self):
        try:
            domain = next(self.domains)
        except StopIteration:
            domain_generator.stop()
            ioloop.IOLoop.instance().stop()
        else:
            fetch_domain(domain)

domain_generator = ioloop.PeriodicCallback(DomainFetcher(domains), 500)
domain_generator.start()

Answer 1

请注意，tornado.ioloop.PeriodicCallback需要一个周期时间以整数毫秒为单位，而HTTPRequest 对象配置有 < code>connect_timeout 和/或 request_timeout 浮点秒 (参见文档）。

“当从点击到响应的延迟小于 100 毫秒时，浏览互联网的用户会感觉响应是“即时”的”(来自维基百科）参见此 ServerFault 问题针对正常延迟 价值观。

IOError: Stream is close 被有效地引发，以通知您连接超时而未完成，或者更准确地说，您在尚未打开的管道上手动调用了回调。这很好，因为延迟大于 1 并不是异常。 100毫秒；如果您希望提取能够可靠地完成，则应该提高该值。

一旦您将超时设置为合理的值，请考虑将您的提取包装在 try/ except 重试循环中，因为这是您可能会在生产中发生的正常异常。请注意设置重试限制！

---

既然您使用的是异步框架，为什么不让它自己处理异步回调，而不是按固定时间间隔运行所述回调？ Epoll/kqueue 非常高效并且受此框架支持。

import ioloop

def handle_request(response):
    if response.error:
        print "Error:", response.error
    else:
        print response.body
    ioloop.IOLoop.instance().stop()

http_client = httpclient.AsyncHTTPClient()
http_client.fetch("http://www.google.com/", handle_request)
ioloop.IOLoop.instance().start()

^ 从文档中逐字复制。

如果您走这条路，唯一的问题是对请求队列进行编码，以便强制执行最大打开连接数。否则，在进行严重的刮擦时，您可能会遇到竞争状况。

自从我自己接触 Tornado 以来已经有大约一年了，所以如果这个回复中有不准确的地方，请告诉我，我会修改。

Answer 2

看起来你正在写类似网络爬虫的东西。您的问题是直接由超时引起的，但在深层，与龙卷风中的并行模式有关。

当然，tornado 中的 AsyncHTTPClient 可以自动对请求进行排队。实际上，AsyncHTTPClient 会批量发送 10 个请求（默认情况下），并阻塞等待结果，然后发送下一批。批次内的请求是非阻塞的并且并行处理，但批次之间是阻塞的。并且每个请求的回调不会在该请求完成后立即调用，而是在该批请求完成后调用 10 个回调。

回到你的问题，你不需要使用 ioloop.PeriodicCallback 来增量地发出请求，因为龙卷风中的 AsyncHTTPClient 可以自动对请求进行排队。您可以一次性分配所有请求，让 AsyncHTTPClient 来调度请求。

但问题来了，等待队列中的请求仍然消耗超时时间！因为请求在批次之间被阻塞。后面的请求只是在这里阻塞，然后批量发送，而不是将它们放入特殊的就绪队列中，并在响应到达后发送新的请求。

因此，如果调度的请求较多，默认的超时设置为 20 秒就太短了。如果您只是做一个演示，您可以直接将超时设置为float('inf')。如果做一些严肃的事情，你必须使用 try/ except 重试循环。

您可以从 tornado/httpclient.py< 找到如何设置超时/code>，在这里引用。

connect_timeout：初始连接超时（以秒为单位）
request_timeout：整个请求的超时时间（以秒为单位）

最后，我编写了一个简单的程序，使用 AsyncHTTPClient 从 ZJU 在线评审系统获取数千个页面。你可以尝试一下，然后重写你的爬虫。在我的网络上，它可以在 2 分钟内获取 2800 个页面。非常好的结果，比串行读取快 10 倍（完全匹配并行大小）。

#!/usr/bin/env python
from tornado.httpclient import AsyncHTTPClient, HTTPRequest
from tornado.ioloop import IOLoop

baseUrl = 'http://acm.zju.edu.cn/onlinejudge/showProblem.do?problemCode='

start = 1001
end = 3800
count = end - start
done = 0

client = AsyncHTTPClient()

def onResponse(response):
    if response.error:
        print('Error: %s' % response.error)
    else:
        global done
        done += 1
        #It is comment out here, you could uncomment it and watch something interest, that len(client.queue) is reduce 10 by 10.
        #print('Queue length %s, Active client count %s, Max active clients limit %s' % (len(client.queue), len(client.active), client.max_clients))
        print('Received %s, Content length %s, Done %s' % (response.effective_url[-4:], len(response.body), done))
        if(done == count):
            IOLoop.instance().stop()

for i in range (start, end):
    request = HTTPRequest(baseUrl + str(i), connect_timeout=float('inf'), request_timeout=float('inf'))
    client.fetch(request, onResponse)
    print('Generated %s' % i)

IOLoop.instance().start()

额外：

如果您有大量页面需要获取，并且您是那种追求最佳性能的人，那么您可以看看 Twisted。我用 Twisted 编写了一个相同的程序，并将其粘贴到我的 Gist 上。其结果非常棒：40 秒内获取 2800 个页面。