如何在不阻塞系统的情况下给出延迟的 AMP 回复？

Question

（我非常乐意接受有关更好标题的建议。）

我正在使用 Twisted 上的 AMP 协议来创建一个调度程序，向其代理提供作业。代理从调度程序中提取作业，因此调度程序是 AMP 服务器，代理作为客户端进行连接。

这个想法是让代理进行连接，从（内部调度程序）作业队列的顶部拾取作业，然后继续执行它。但是，不能保证该队列始终非空。因此，我希望利用扭曲的延迟机制，以便在调度程序设法从队列中弹出作业时，在代理端简单地进行延迟触发。

不过，在调度程序端实现这一点被证明有点棘手。 AMP 的工作方式是为代理可以发送的每个（由我预定义的）命令分配一个函数，该函数获取该命令具有的所有参数并返回其返回的所有值的字典。这意味着我需要在一个函数内完成这一切。通常情况下，这不会是一个问题，但这里扭曲似乎妨碍了我：我需要让函数暂停一会儿，而不暂停扭曲事件循环，从而允许它实际上向队列添加更多作业，所以一个可以弹出。 （这就是我认为通常的 sleep() 不会达到预期效果的原因。）更重要的是，这意味着我想不出一种方法来使用一些扭曲的功能，例如 < code>deferToThread()，因为我必须在单独的函数中处理该结果（并且只能访问它们），我将其分配为该 deferred 的回调，所以我不知道要返回什么启动单独的线程并分配其回调后的 AMP 响应程序函数。这更清楚地说明了我的意思：（

def assignJob(agentID):
    # We expect the agentID, so we can store who we've given a job to.

    # Get a job without blocking even if the queue is originally empty.
    job = None
    while job is None:
        try:
            job = jobqueue.pop(0)
        except IndexError:
            # Imagine getJob simply tries to get a job every 5 seconds
            # (using sleep() safely because it's in a separate thread)
            # until it eventually gets one, which it returns
            d = deferToThread(getJob)

            # We would then need to have a separate function
            # , e.g. jobReturn() pick up the firing deferred and do
            # something with the result...
            d.addCallback(jobReturn)   

    # But if we do... We don't (necessarily) have a job to return here
    # because for all we know, the deferred from that thread hasn't even
    # fired yet.
    return {'job': ???}

这显然不是该函数的实际完整代码 - 首先，它是所需的 amp.AMP 子类的方法。）

reactor 方法callInThread() 起初似乎也很有用（因为它不返回延迟），但它没有提供一种方法来获取它执行的可调用函数的返回值（据我所知）可以看到）并且，即使确实如此，这意味着等待线程完成，这会阻塞此方法很长时间，这使得使用单独的线程毫无意义。

那么，在我找到工作之前，如何阻止此方法，而不是整个 Twisted 事件循环，或者如何在其立即响应程序方法之外返回 AMP 回复？

Answer 1

您可能错过的一件事是，AMP 响应程序方法本身也允许返回 Deferred（在 AMP API 文档）。只要 Deferred 最终使用与命令的响应定义相匹配的字典来触发，一切都会正常工作。

也有些相关，如果您想避免使用线程，您可能需要查看 twisted.internet.defer.DeferredQueue，一种原生了解 Deferred 的队列数据结构。