后台模式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()

我想要做的关键是在后台线程中运行服务器.我希望用线程而不是通过多处理/队列来做这个,因为我已经为我的应用程序提供了一层消息,我想避免两个.我提出这个问题是因为我已经可以在一个单独的过程中看到如何做到这一点,但我想知道的是如何在一个线程中完成它,或者如果可以的话.或者,如果可能还有一些其他模式我可以使用它完成同样的事情,比如编写我自己的reactor.run方法.谢谢你的帮助.:)

Answer 1

我想要做的关键是在后台线程中运行服务器.

但是,你不解释为什么这是关键.通常,"使用线程"之类的东西是实现细节.也许线程是合适的,也许不是,但实际目标在这一点上是不可知的.你的目标是什么？同时处理多个客户端？要同时处理来自其他来源(例如,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是如何定义的.我想象它以某种方式与网络接口.你的代码版本会让它接收消息并缓冲它们,直到它们被你的循环从缓冲区中删除进行处理.这个版本可能没有缓冲区,而是只要在消息到达时调用messageReceived传递给对象的方法start.如果需要,您仍然可以通过将其添加到messageReceived方法中来添加某种缓冲.

• 现在有一个电话reactor.run会阻止.您可能会将此代码编写为twistd插件或.tac文件,在这种情况下,您不会直接负责启动reactor.但是,有人必须启动反应堆,或者Twisted的大多数API都不会执行任何操作. reactor.run当然,直到有人打电话为止reactor.stop.

• 这种方法没有使用线程.Twisted的协同多任务处理并发意味着你可以同时做多件事,只要你注意合作(这通常意味着偶尔回到反应堆).

• doSomeOtherLogic调用函数的确切时间会稍微改变,因为"我刚刚处理了一条消息","缓冲区现在为空"没有概念.您可以更改此设置,以便每秒调用一次,或者每N次消息后调用一次,或者任何适当的安装.

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

from Queue import Queue
from threading import Thread

class MessageQueuer(object):
    def __init__(self, queue):
        self.queue = queue

    def messageReceived(self, message):
        self.queue.put(message)

def main():
    queue = Queue()
    twistedServer = ...
    twistedServer.start(MessageQueuer(queue))
    Thread(target=reactor.run, args=(False,)).start()

    while True:
        message = queue.get()
        response = handleMessage(message)
        reactor.callFromThread(twistedServer.sendResponse, response)

main()

这个版本假设一个twistedServer工作方式类似,但使用一个线程让你有while True:循环.注意:

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

• MessageQueuer具有相同的接口MessageReverser,只是它的实现messageReceived是不同的.它使用线程安全的Queue对象在反应器线程(将在其中调用它)与while True:运行循环的主线程之间进行通信.

• 您必须使用reactor.callFromThread将消息发送回reactor线程(假设twistedServer.sendResponse实际上是基于Twisted API).Twisted API通常不是线程安全的,必须在reactor线程中调用.这reactor.callFromThread对你有什么帮助.

• 你想要实现一些方法来停止循环和反应堆.在你打电话之前,python进程不会干净地退出reactor.stop.

请注意,虽然线程版本为您提供了熟悉的,所需的while True循环,但它实际上并没有比非线程版本做得更好.它更复杂.因此,请考虑您是否确实需要线程,或者它们是否仅仅是可以与其他东西交换的实现技术.