python - 如何在不阻塞系统的情况下提供延迟的 AMP 回复？

Question

(我非常愿意接受更好标题的建议。)

我在 Twisted 上使用 AMP 协议(protocol)来创建一个调度程序，将作业提供给它的代理。代理从调度程序中提取作业，因此调度程序是 AMP 服务器，代理作为客户端连接。

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

不过，事实证明，在调度程序端实现这一点有点棘手。 AMP 的工作方式是为代理可以发送的每个(由我预定义的)命令分配一个函数，该函数采用命令具有的所有参数并返回它返回的所有值的字典。这意味着我需要在一个函数中完成这一切。通常情况下，这不是问题，但这里 twisted 似乎妨碍了我:我需要让函数暂停一下，而不暂停 twisted 事件循环，从而允许它实际向队列中添加更多作业，所以一个可以弹出。 (这就是我不认为通常的 sleep() 会产生预期效果的原因。)更重要的是，这意味着我想不出一种方法来使用一些扭曲的功能，例如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 docs 中搜索 may also return Deferreds)。只要 Deferred 最终使用与命令的响应定义相匹配的字典触发，一切都会正常进行。

还有点相关，如果你想避免使用线程，你可能想看看 twisted.internet.defer.DeferredQueue ，一个队列数据结构，它本身就知道 Deferreds。