我对twisted很陌生,我正在尝试制作一个异步客户端,它可以获取一些url并将结果保存到每个url的不同文件中。当我用有限数量的服务器(比如10台)运行程序时,reactor循环正确结束,程序终止。但是,当我使用Alexa top 2500运行该程序时,该程序开始获取URL,但随后没有终止。我已经设置了超时,但它不起作用,我相信一定有一些打开的套接字不会触发任何回调,无论是错误还是成功。我的目标是,一旦程序获取了页面或每个连接的超时时间到期,程序就必须终止并关闭所有活动的文件描述符。
很抱歉,复制和粘贴时没有保留代码缩进,现在我已经检查并修复了它。举个例子,代码是最低限度的,请注意,我的问题是,当我启动程序时,有大量的站点需要爬网,反应器没有停止。
#!/usr/bin/env python
from pprint import pformat
from twisted.internet import reactor
import twisted.internet.defer
import sys
from twisted.internet.protocol import Protocol
from twisted.web.client import Agent
from twisted.web.http_headers import Headers
class PrinterClient(Protocol):
def __init__(self, whenFinished, output):
self.whenFinished = whenFinished
self.output = output
def dataReceived(self, bytes):
#print '##### Received #####n%s' % (bytes,)
self.output.write('%s' % (bytes,))
def connectionLost(self, reason):
print 'Finished:', reason.getErrorMessage()
self.output.write('Finished: %s n'%(reason.getErrorMessage()))
self.output.write('#########end########%sn'%(reason.getErrorMessage()))
self.whenFinished.callback(None)
def handleResponse(r, output, url):
output.write('############start############n')
output.write('%sn'%(url))
#print "version=%sncode=%snphrase='%s'" % (r.version, r.code, r.phrase)
output.write("version=%sncode=%snphrase='%s'"
%(r.version, r.code, r.phrase))
for k, v in r.headers.getAllRawHeaders():
#print "%s: %s" % (k, 'n '.join(v))
output.write("%s: %sn" % (k, 'n '.join(v)))
whenFinished = twisted.internet.defer.Deferred()
r.deliverBody(PrinterClient(whenFinished, output))
return whenFinished
def handleError(reason):
print reason
#reason.printTraceback()
#reactor.stop()
def getPage(url, output):
print "Requesting %s" % (url,)
d = Agent(reactor).request('GET',
url,
Headers({'User-Agent': ['Mozilla/4.0 (Windows XP 5.1) Java/1.6.0_26']}),
None)
d._connectTimeout = 10
d.addCallback(handleResponse, output, url)
d.addErrback(handleError)
return d
if __name__ == '__main__':
semaphore = twisted.internet.defer.DeferredSemaphore(500)
dl = list()
ipset = set()
queryset = set(['http://www.google.com','http://www.google1.com','http://www.google2.com', "up to 2500 sites"])
filemap = {}
for q in queryset:
fpos = q.split('http://')[1].split(':')[0]
dl.append(semaphore.run(getPage, q, filemap[fpos]))
dl = twisted.internet.defer.DeferredList(dl)
dl.addCallbacks(lambda x: reactor.stop(), handleError)
reactor.run()
for k in filemap:
filemap[k].close()
谢谢。Jeppo
您的超时代码至少有两个问题。
首先,您设置的唯一超时是_connectTimeout
,并将其设置在从Agent.request
返回的Deferred
上。这是一个没有意义的属性,Agent
实现中和Twisted的任何部分都不会尊重它。我认为你是想在Agent
实例上设置这个属性,因为它会产生一些影响。然而,它是一个私有属性,不适合您直接与之交互。相反,应该将connectTimeout=10
传递给Agent
初始值设定项。
其次,此超时仅影响TCP连接设置超时。将其设置为10
意味着,如果无法在10秒内建立到特定URL的HTTP服务器的TCP连接,则请求尝试将失败,并出现超时错误。但是,如果在不到10秒内成功建立了连接,则超时就没有进一步的意义了。如果服务器需要10个小时才能向您发送响应,Agent
将在那里等待10个小时。您需要一个额外的超时,一个完整的请求超时。
这是使用reactor.callLater
和可能的Deferred.cancel
分别实现的。例如,
...
d = agent.request(...)
timeoutCall = reactor.callLater(60, d.cancel)
def completed(passthrough):
if timeoutCall.active():
timeoutCall.cancel()
return passthrough
d.addBoth(completed)
...