填充传入消息队列并清空传出消息队列的后台Twisted服务器的模式？

Question

我想这样做：

twistedServer.start() # This would be a nonblocking call

while True:
   while twistedServer.haveMessage():
      message = twistedServer.getMessage()
      response = handleMessage(message)
      twistedServer.sendResponse(response)
   doSomeOtherLogic()

我想做的关键是在后台线程中运行服务器。我希望用线程而不是通过多处理/队列来实现这一点，因为我已经为我的应用程序提供了一层消息传递，我希望避免两层消息传递。我之所以提出这个问题，是因为我已经知道如何在一个单独的进程中实现这一点，但我想知道的是如何在一个线程中执行，或者如果可以的话。也许我可以用我自己的模式来完成这件事反应器.运行方法。谢谢你的帮助。 ：）

Answer 1

The key thing I want to do is run the server in a background thread.

你不能解释为什么这是关键。通常，“使用线程”之类的东西是实现细节。也许线程是合适的，也许不是，但实际的目标在这一点上是不可知的。你的目标是什么？同时处理多个客户机？要同时处理来自另一个源（例如，web服务器）的事件这类消息？如果不知道最终目标，就无法知道我建议的实施策略是否有效。在

考虑到这一点，这里有两种可能性。在

首先，你可以忘记线程。这需要将上面的事件处理逻辑定义为仅事件处理部分。尝试获取事件的部分将被委托给应用程序的另一部分，可能最终基于某个reactor api（例如，您可以设置一个TCP服务器来接收消息并将其转换为您正在处理的事件，在这种情况下，你可以先打电话给reactor.listenTCP某种程度上）。在

因此，您的示例可能会变成这样（通过一些附加的特殊性来增加指导价值）：

from twisted.internet import reactor

class MessageReverser(object):
    """
    Accept messages, reverse them, and send them onwards.
    """
    def __init__(self, server):
        self.server = server

    def messageReceived(self, message):
        """
        Callback invoked whenever a message is received.  This implementation
        will reverse and re-send the message.
        """
        self.server.sendMessage(message[::-1])
        doSomeOtherLogic()

def main():
    twistedServer = ...
    twistedServer.start(MessageReverser(twistedServer))
    reactor.run()

main()

关于这个例子需要注意的几点：

• 我不确定您的twistedServer是如何定义的。我在想象它以某种方式与网络连接。您的代码版本会让它接收消息并对其进行缓冲，直到循环将它们从缓冲区中移除以进行处理。此版本可能没有缓冲区，而是在消息到达时立即调用传递给start的对象的messageReceived方法。如果需要，还可以添加某种类型的缓冲区，方法是将其放入messageReceived方法中。

• 现在有一个对reactor.run的调用，该调用将阻止。您可以将此代码写成twistd插件或.tac文件，在这种情况下，您将不直接负责启动反应堆。但是，必须有人启动反应堆，否则来自Twisted的大多数api不会做任何事情。reactor.run当然，直到有人调用reactor.stop。

• 此方法没有使用线程。Twisted的协同多任务处理方法意味着你仍然可以同时做多件事情，只要你注意合作（这通常意味着偶尔返回反应堆）。

• 调用doSomeOtherLogic函数的确切时间稍有变化，因为与“我刚刚处理了一条消息”没有“缓冲区暂时为空”的概念。您可以更改此设置，以便每隔一秒或在每N条消息之后或任何适当的方式安装该函数。

第二种可能是真正使用线程。这看起来可能与前面的示例非常相似，但是您应该在另一个线程中调用reactor.run，而不是主线程。例如

^{pr2}$

这个版本假设一个twistedServer，它的工作原理类似，但是使用一个线程让您拥有while True:循环。注：

• 如果使用线程，则必须调用reactor.run(False)，以防止Twisted尝试安装任何信号处理程序，Python只允许在主线程中安装这些处理程序。这意味着Ctrl-C处理将被禁用，reactor.spawnProcess将无法可靠地工作。

• MessageQueuer与MessageReverser具有相同的接口，只是messageReceived的实现不同。它使用threadsafe Queue对象在reactor线程（它将被调用）和运行while True:循环的主线程之间进行通信。

• 必须使用reactor.callFromThread将消息发送回reactor线程（假设twistedServer.sendResponse实际上基于Twisted api）。Twisted api通常不是线程安全的，必须在reactor线程中调用。这就是reactor.callFromThread为你做的。

• 你需要实现某种方法来停止回路和反应堆，有人推测。在调用reactor.stop之后，python进程才会干净地退出。

注意，虽然线程版本提供了熟悉的、期望的while True循环，但实际上它并没有比非线程版本做得更好。只是更复杂。所以，考虑一下你是否真的需要线程，或者它们仅仅是一种可以交换其他东西的实现技术。在