Python Twisted deferToThread函数未按预期工作

Question

我这里的目标是创建一个简单的服务器来处理TCP消息，比如“processmystring”，它发送“mystring”由一个非常耗时的操作（这里是slowFunction）来处理。这里我通过deferToThread调用这个函数，但似乎什么也没有发生：延迟的回调消息在任何地方都不会显示（断点显示它从未被调用），并且函数中的打印也不会显示（断点显示它从未被调用过）

from twisted import protocols
from twisted.protocols import basic
from twisted.internet import threads, protocol, reactor
from twisted.application import service, internet
import re
import time

def slowFunction(arg):
    print "starting process"
    time.sleep(1)
    print "processed "+arg

class MySimpleServer(basic.LineReceiver):

    PROCESS_COMMAND = "process (.*)" #re pattern
    processFunction = slowFunction
    clients = []

    def connectionMade(self):
        print "Client connected"
        MySimpleServer.clients.append(self)

    def connectionLost(self, reason):
        print "Client gone"
        MySimpleServer.clients.remove(self)

    def onProcessDone(self):
        self.message("Process done")

    def onError(self):
        self.message("Process fail with error")

    def lineReceived(self, line):
        processArgumentResult = re.search(MySimpleServer.PROCESS_COMMAND, line)
        if not processArgumentResult == None:
            processArgument = processArgumentResult.groups()[0] 
            deferred = threads.deferToThread(MySimpleServer.processFunction, processArgument)
            deferred.addCallback(self.onProcessDone)
            deferred.addErrback(self.onError)
            self.message("processing your request")
        else:
            print "Unknown message line: "+line

    def message(self, message):
        self.transport.write(message + '\n')

if __name__ == '__main__':
    factory = protocol.ServerFactory()
    factory.protocol = MySimpleServer
    factory.client = []

    reactor.listenTCP(8000, factory)
    reactor.run()

Answer 1

我得到了twisted irc的人的帮助

要点是：回调（onProcessDone和onError）应该接受一个结果参数，而deferToThread调用的函数将接收self作为参数（它应该是MySimpleServer类的方法之一）。在

最后的代码是：

from twisted import protocols
from twisted.protocols import basic
from twisted.internet import threads, protocol, reactor
from twisted.application import service, internet
import re
import time

def slowFunction(arg):
    print "starting process"
    time.sleep(20)
    print "processed "+arg

class MySimpleServer(basic.LineReceiver):

    PROCESS_COMMAND = "process (.*)" #re pattern
    clients = []

    def connectionMade(self):
        print "Client connected"
        MySimpleServer.clients.append(self)

    def connectionLost(self, reason):
        print "Client gone"
        MySimpleServer.clients.remove(self)

    def onProcessDone(self, result):
        self.message("Process done")

    def onError(self, result):
        self.message("Process fail with error")

    def processFunction(self, processArgument):
        slowFunction(processArgument)

    def lineReceived(self, line):
        processArgumentResult = re.search(MySimpleServer.PROCESS_COMMAND, line)
        if not processArgumentResult == None:
            processArgument = processArgumentResult.groups()[0] 
            deferred = threads.deferToThread(self.processFunction, processArgument)
            deferred.addCallback(self.onProcessDone)
            deferred.addErrback(self.onError)
            self.message("processing your request")
        else:
            print "Unknown message line: "+line

    def message(self, message):
        self.transport.write(message + '\n')

if __name__ == '__main__':
    factory = protocol.ServerFactory()
    factory.protocol = MySimpleServer
    factory.client = []

    reactor.listenTCP(8000, factory)
    reactor.run()

Answer 2

另一种方法是使用staticmethod；这是它的合法用途。在

class MySimpleServer(basic.LineReceiver):
    processFunction = staticmethod(slowFunction)