正确使用线程化双绞线中的API

Question

我使用的是一个API使用threading.Thread的服务，我想在其中使用Deferreds。

如果我像一个标准python模块一样运行它，我没有任何问题。类似于：

from twisted.internet import reactor
from outside import ServiceUsingThreadingAndDeferred

service = ServiceUsingThreadingAndDeferred()

reactor.listenTCP(port, protocol_factory)

service.start()

reactor.run()

如果我使用扭曲的.tac运行以下-y，则服务根本无法工作：

from twisted.application import internet, service
from outside import ServiceUsingThreadingAndDeferred

service = ServiceUsingThreadingAndDeferred()

# Application set-up
application = service.Application("appName")
my_server = internet.TCPServer(port, protocol_factory)
my_server.setServiceParent(application)

service.start()

我认为第二种情况下的问题是主反应堆线程本身并不生成服务线程，但是我不明白为什么(如果)在第一种情况下发生这种情况.我使用callLater作为解决-successfully的方法：

from twisted.application import internet, service
from outside import ServiceUsingThreadingAndDeferred
from twisted.internet import reactor

service = ServiceUsingThreadingAndDeferred()

# Application set-up
application = service.Application("appName")
my_server = internet.TCPServer(port, protocol_factory)
my_server.setServiceParent(application)

reactor.callLater(1, service.start)

但我不知道这是不是解决问题的正确方法。你有什么建议吗？

Answer 1

这个类错误地使用了Twisted的线程API：

class ServiceUsingThreadingAndDeferred():
    def __init__(self):
        pass

    def start(self):
        print "3rd party API service starting..."
        self.run_as_thread()

    def run_as_thread(self, *args, **kwargs):
        t = threading.Thread(target=self.run_forever, args=args, kwargs=kwargs)
        t.daemon = True
        t.start()

    def run_forever(self):
        while 1:
            print "Doing something remote..."
            time.sleep(1)
            now = time.time()
            if 1 > now % 5 >= 0:
                self.defer_activity()

    def defer_activity(self):
        threads.deferToThread(self._activity)

ServiceUsingThreadingAndDeferred.run_forever运行在一个非反应堆线程中。它叫defer_activity，它叫threads.deferToThread。不允许在非反应堆线程中调用threads.deferToThread。大约有一个Twisted可以安全地调用一个非反应堆线程：reactor.callFromThread (它将调用调度到其参数以在反应堆线程中运行)。

working.tac也做了同样的事情，但幸运的是，在Twisted的某些版本上似乎也是如此。它依赖于在非反应堆线程中调用threads.deferToThread与如何实现callLater交互而产生的未定义行为。不能保证它完全工作，也不能保证这种行为可以跨Twisted版本或平台进行移植。

如果要从非反应堆线程中使用反应堆的线程池，则需要编写如下内容：

from twisted.internet.threads import (
    blockingCallFromThread,
    deferToThread,
)

d = blockingCallFromThread(reactor, lambda: deferToThread(self._activity))

但是，您也不能在非反应堆线程中使用d的任何方法(deferToThread返回的Deferred )。

最有可能的是，如果有可能的话，你应该重写ServiceUsingThreadingAndDeferred的逻辑，这样它就能与反应堆兼容，然后你就可以避免所有这些诡计了。