龙卷风异步HTTP客户端抓取异常

Question

我正在使用tornado.httpclient.AsyncHTTPClient.fetch从列表中获取域。当我以较大的间隔（例如500）将域放入fetch时，一切都很好，但是当我将interval减少到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

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

^{pr2}$

Answer 1

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

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

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

但问题是等待队列中的请求仍然占用超时时间！因为请求在批处理之间被阻塞。稍后的请求在这里简单地阻塞，然后一批一批地发送，而不是将它们放入一个特殊的就绪队列中，在响应到达后发送一个新的请求。在

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

您可以在这里找到如何从^{}设置超时。在

connect_timeout: Timeout for initial connection in seconds
request_timeout: Timeout for entire request in seconds

最后，我编写了一个简单的程序，使用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页。在

Answer 2

注意，tornado.ioloop.PeriodicCallbacktakes a cycle time in integer ms，而^{}对象配置了一个connect_timeout和/或一个request_timeout的float秒（see doc）。在

“浏览互联网的用户觉得，当从单击到响应的延迟小于100毫秒时，响应是“即时的”（from wikipedia）请参见this ServerFault question for normal latency values。在

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

一旦将超时设置为合理的值，请考虑将获取打包到try/except retry循环中，因为这是一个正常的异常，您可以预期在生产中发生。小心设置重试限制！在

---

既然您使用的是异步框架，为什么不让它自己处理异步回调，而不是在固定的时间间隔内运行所述回调呢？Epoll/kqueue are efficient and supported by this framework.

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()

^逐字复制from the doc。在

如果你走这条路，唯一的问题就是对你的请求队列进行编码，这样你就有了最大限度的开放连接。否则，当你做严重的刮擦时，你很可能会出现竞速状态。在

自从我自己接触龙卷风已经有大约1年了，所以请告诉我这个回答是否有不准确之处，我会修改的。在