4

我对twisted很陌生，我正在尝试制作一个异步客户端来获取一些url并将结果保存到每个url的不同文件中。当我用有限数量的服务器运行程序时，比如说 10 台，反应器循环正确结束并且程序终止。但是，当我使用例如 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########%s\n'%(reason.getErrorMessage()))
        self.whenFinished.callback(None)

def handleResponse(r, output, url):
    output.write('############start############\n')
    output.write('%s\n'%(url))
    #print "version=%s\ncode=%s\nphrase='%s'" % (r.version, r.code, r.phrase)
    output.write("version=%s\ncode=%s\nphrase='%s'"\
             %(r.version, r.code, r.phrase))
    for k, v in r.headers.getAllRawHeaders():
        #print "%s: %s" % (k, '\n  '.join(v))
        output.write("%s: %s\n" % (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()

谢谢。捷波

4

1 回答 1

8

您的超时代码至少有两个问题。

首先，您设置的唯一超时是_connectTimeout，并且您将其设置在Deferred从Agent.request. 这是一个无意义的属性，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)
...
于 2013-02-28T18:08:53.920 回答