python - 扭曲/执行异步http请求

Question

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

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

我目前拥有的是:

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

http react 器使用 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 检查添加到“主” react 器，以便它们以异步方式执行？ DeferredSemaphore 究竟是如何工作的？

谁能帮我解决这个问题？

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

Answer 1

您不需要多个 react 堆。只需使用同一个 react 器执行所有不同的操作。

如果你正在调用 reactor.stop()，你可能做错了什么，所以让我们摆脱它，并将它全部绑定(bind)到一个函数中(我们可以使用作为回调)；因为它正在进行异步工作，它还应该返回一个延迟，我们将使用您已经在使用的 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)

---

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

在 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() 的情况不太好。您不能使用具有该接口(interface)的任何其他 react 器；您需要使用更新、更 Shiny 的 t.w.c.Agent .在许多方面 Agent 更胜一筹，但当您只需要将原始响应主体作为字符串时，它就不太方便了。

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

这里有一段代码可以替代 getPage，具有大致相同的接口(interface)，但将 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 ，更少混淆；但是，尽早鼓起正确处理 react 器并避免以后不必要的痛苦也是一个不错的主意。