4

我有一个扭曲的反应器来监听传入的数据。我有第二个反应器在特定时间间隔内执行 http 请求，将结果发送到第一个反应器。两者都运行良好。

现在我想把它放在一个反应​​堆中运行，但我不知道如何实现。类似的东西 - 每 60 秒执行一次 http 请求。从第一个监听“主”反应器中以异步方式。

我目前拥有的是：

# main reactor listening for incoming data forever
...
reactor.listenTCP(8123, TCPEventReceiverFactory())

http reactor 用于twisted.internet.defer.DeferredSemaphore()执行多个 http 检查：

# create semaphore to manage the deferreds
semaphore = twisted.internet.defer.DeferredSemaphore(2)

# create a list with all urls to check
dl = list()
# append deferreds to list
for url in self._urls:
    # returns deferred
    dl.append(semaphore.run(self._getPage, url))

# get a DefferedList
dl = twisted.internet.defer.DeferredList(dl)
# add some callbacks for error handling
dl.addCallbacks(lambda x: reactor.stop(), self._handleError)

# start the reactor    
reactor.run()

如何将定时 http 检查添加到“主”反应器，以便它们以异步方式执行？具体是如何DeferredSemaphore工作的？

谁能帮我这个？

【这是一种处理http校验结果的轻量级监控系统。我是 Twisted 和异步编程的新手。我在运行 Python 2.7 的 Xubuntu 12.04]

4

1 回答 1

6

你不需要多个反应器。只需使用同一个反应器执行所有不同的操作。

如果你正在调用reactor.stop()，你可能做错了什么，所以让我们摆脱它，并将它全部绑定到一个函数中（我们可以将其用作回调）；因为它在做异步工作，它也应该返回一个延迟的，我们将使用DeferredList你已经在使用的。

def thing_that_does_http():
    # create semaphore to manage the deferreds
    semaphore = twisted.internet.defer.DeferredSemaphore(2)

    # create a list with all urls to check
    dl = DeferredList()
    # append deferreds to list
    for url in self._urls:
        # returns deferred
        dl.append(semaphore.run(self._getPage, url))

    # get a DefferedList
    dl = twisted.internet.defer.DeferredList(dl)
    # add some callbacks for error handling
    dl.addErrback(self._handleError)
    return dl

“在特定时间间隔内执行x ”的自然方法是循环调用。有了这个回调函数，我们不需要做太多

reactor.listenTCP(8123, TCPEventReceiverFactory())
loop_http = twisted.intertnet.task.LoopingCall(thing_that_does_http)
# run once per minute, starting now.
loop_http.start(60)

反应器LoopingCall将getPage用于自己的目的twisted.internet.reactor，如果您使用不同的反应器，例如，如果您正在进行单元测试，则需要覆盖该默认值。

在 的情况下LoopingCall，很简单，在构造之后（但在调用它的start()方法之前），设置它的clock属性：

from twisted.internet.task import Clock
fake_reactor = Clock()
loop_http.clock = fake_reactor
fake_reactor.advance(120)  # move time forward two minutes...

不幸的是，情况getPage()不太好。您不能使用具有该接口的任何其他反应器；您需要使用更新的、更闪亮的t.w.c.Agent. 在许多方面Agent都是优越的，但是当您只想要原始响应主体作为字符串时，它就不那么方便了。

除了要求将显式反应器传递给其构造函数之外，它更多的是对请求/响应周期的细粒度控制，而不是 getPage 提供的便利。因此，它主要是根据Producers 和Protocols 来实现的。在前者的情况下，我们可以传递一个方便的助手，FileBodyProducer以最小的麻烦发送请求体；在后者中，我们需要一个简单的协议来缓冲所有数据块，直到我们得到所有数据。

这是一段可以替换的代码getPage，具有大致相同的接口，但将 的实例Agent作为第一个参数

from cStringIO import StringIO
from twisted.internet.defer import Deferred
from twisted.internet.protocol import Protocol
from twisted.web.client import ResponseDone
from twisted.web.client import FileBodyProducer


class GetPageProtocol(Protocol):
    def __init__(self):
        self.deferred = Deferred()
        self.data = []

    def dataReceived(self, data):
        self.data.append(data)

    def connectionLost(self, reason):
        reason.trap(ResponseDone)
        data = ''.join(self.data)
        del self.data
        self.deferred.callback(data)


def agentGetPage(agent, url,
                 method="GET",
                 headers=None,
                 postdata=None):
    if postdata is not None:
        bodyProducer = FileBodyProducer(StringIO(postdata))
    else:
        bodyProducer = None

    def _getPageResponded(response):
        if response.length != 0:
            proto = GetPageProtocol()
            response.deliverBody(proto)
            return proto.deferred
        else:
            return None

    d = agent.request(method, url, headers, bodyProducer)
    d.addCallback(_getPageResponded)
    return d

在单元测试中，它看起来有点像：

from twisted.test.proto_helpers import MemoryReactor
from twisted.web.client import Agent
fake_reactor = MemoryReactor()
agent = Agent(fake_reactor)
d = agentGetPage(agent, "http://example.com")

assert fake_reactor.tcpClients  # or some such, exercise the code by manipulating the reactor

编辑：我最初想略过这个给ectomorph，不要混淆；但是，尽早正确处理反应器也是一个不错的主意，然后避免不必要的痛苦。

于 2013-05-14T12:44:13.700 回答