对 Python Twisted 透视代理进行远程调用排队？

Question

Twisted（对于 python）的优势在于它的异步框架（我认为）。我编写了一个图像处理服务器，它通过 Perspective Broker 接收请求。只要我一次给它提供少于几百张图像，它就可以很好地工作。然而，有时它几乎同时出现数百张图像。因为它试图同时处理所有这些，所以服务器崩溃了。

作为一个解决方案，我想在服务器上对 Remote_calls 进行排队，以便它一次仅处理约 100 个图像。看起来这可能是 Twisted 已经做的事情，但我似乎找不到它。关于如何开始实施这个有什么想法吗？方向正确的一点？谢谢！

Answer 1

一个可能对此有所帮助的现成选项是twisted.internet.defer.DeferredSemaphore。这是普通（计数）信号量的异步版本，如果您做过很多线程编程，您可能已经知道了。

（计数）信号量很像互斥锁（锁）。但是，如果互斥锁在相应的释放之前只能获取一次，则可以将（计数）信号量配置为允许在需要任何相应的释放之前成功进行任意（但指定的）次数的获取。

下面是使用 DeferredSemaphore 运行十个异步操作的示例，但一次最多运行其中三个：

from twisted.internet.defer import DeferredSemaphore, gatherResults
from twisted.internet.task import deferLater
from twisted.internet import reactor


def async(n):
    print 'Starting job', n
    d = deferLater(reactor, n, lambda: None)
    def cbFinished(ignored):
        print 'Finishing job', n
    d.addCallback(cbFinished)
    return d


def main():
    sem = DeferredSemaphore(3)

    jobs = []
    for i in range(10):
        jobs.append(sem.run(async, i))

    d = gatherResults(jobs)
    d.addCallback(lambda ignored: reactor.stop())
    reactor.run()


if __name__ == '__main__':
    main()

DeferredSemaphore 还具有显式的 acquire 和release 方法，但是 run 方法非常方便，几乎总是您想要的。它调用 acquire 方法，该方法返回一个 Deferred。对于第一个 Deferred，它添加了一个回调，该回调调用您传入的函数（以及任何位置或关键字参数）。如果该函数返回一个 Deferred，则向第二个 Deferred 添加一个调用 release 方法的回调。

同步情况也可以通过立即调用release来处理。还可以通过允许错误传播来处理错误，但确保完成必要的释放以使 DeferredSemaphore 保持一致状态。传递给 run 的函数结果（或其返回的 Deferred 的结果）将成为 返回的 Deferred 的结果运行。

另一种可能的方法可能基于DeferredQueue 和cooperate。 DeferredQueue 与普通队列基本相似，但其 get 方法返回一个 Deferred。如果调用时队列中没有项目，则在添加项目之前，Deferred 将不会触发。

下面是一个示例：

from random import randrange

from twisted.internet.defer import DeferredQueue
from twisted.internet.task import deferLater, cooperate
from twisted.internet import reactor


def async(n):
    print 'Starting job', n
    d = deferLater(reactor, n, lambda: None)
    def cbFinished(ignored):
        print 'Finishing job', n
    d.addCallback(cbFinished)
    return d


def assign(jobs):
    # Create new jobs to be processed
    jobs.put(randrange(10))
    reactor.callLater(randrange(10), assign, jobs)


def worker(jobs):
    while True:
        yield jobs.get().addCallback(async)


def main():
    jobs = DeferredQueue()

    for i in range(10):
        jobs.put(i)

    assign(jobs)

    for i in range(3):
        cooperate(worker(jobs))

    reactor.run()


if __name__ == '__main__':
    main()

请注意，async 辅助函数与第一个示例中的函数相同。不过，这一次，还有一个 worker 函数，它显式地将作业从 DeferredQueue 中拉出，并使用 async 处理它们（通过添加 >async 作为对 get 返回的 Deferred 的回调）。 worker 生成器由 cooperate 驱动，它在每次 Deferred 产生触发后迭代一次。然后，主循环启动其中三个工作生成器，以便在任何给定时间都将进行三个作业。

这种方法比 DeferredSemaphore 方法涉及更多的代码，但有一些可能有趣的好处。首先，cooperate 返回一个 CooperativeTask 实例，该实例具有诸如 pause、resume 等有用的方法。此外，分配给同一合作者的所有作业将在调度中相互合作，以免事件循环过载（这就是 API 名称的由来）。在 DeferredQueue 方面，还可以对等待处理的项目数量设置限制，这样您就可以避免服务器完全超载（例如，如果您的图像处理器卡住并停止完成任务）。如果调用 put 的代码处理队列溢出异常，您可以将此作为压力来尝试停止接受新作业（可能将它们分流到另一台服务器，或警告管理员）。使用 DeferredSemaphore 执行类似的操作有点棘手，因为无法限制等待获取信号量的作业数量。

Answer 2

您可能还喜欢我写的 txRDQ（可调整大小的调度队列）。谷歌一下，它在 LaunchPad 上的 tx 集合中。抱歉，我没有更多时间回复 - 即将上台。

特里