从另一个线程调用一个扭曲的协议方法

Question

我已经用Python做了一个家庭安全程序，它使用Raspberry Pi的GPIO来感应移动并启动警笛。用户使用NFC标签激活/停用系统，也可以连接到树莓派。

为此，我需要不间断地检查nfc标签，同时不断检查传感器的运动是否阻塞。我需要更多并行的东西来做，但我认为这两点足以说明我的观点。

现在我使用的线程，我开始/停止像这样 - Stopping a thread after a certain amount of time - 我不知道这是否是最佳的方式，但到目前为止系统工作正常。

现在我想扩展它的功能来通过websockets提供通知。我发现，这可以用双绞线来实现，但我很困惑..

下面是我如何试图做到这一点的例子代码：

from twisted.internet import reactor 
from autobahn.websocket import WebSocketServerFactory, \ 
           WebSocketServerProtocol, \ 
           listenWS 


def thread1(stop_event): 
    while(not stop_event.is_set()): 
     stop_event.wait(4) 
     print "checking sensor" 
     # sensor_state = GPIO.input(11) 
     if sensor_state == 1: 
      # how can I call send_m("sensor detected movement") #<--- 
      t1_stop_event.set() 

t1_stop_event = Event() 
t1 = Thread(target=thread1, args=(t1_stop_event,)) 

class EchoServerProtocol(WebSocketServerProtocol): 
    def onMessage(self, msg, binary): 
    print "received: "+msg 
    print "stopping thread1" 
    t1_stop_event.set() 

    def send_m(self, msg): 
    self.sendMessage(msg) 

if __name__ == '__main__': 
    t1.start() 
    factory = WebSocketServerFactory("ws://localhost:9000") 
    factory.protocol = EchoServerProtocol 
    listenWS(factory) 
    reactor.run() 

所以，我怎么能叫的发送方法服务器协议从一个线程像thread1？

Answer 1

通常情况下，关于线程和Twisted的问题的答案是“不使用线程”。

您在此处启动线程的原因似乎是，您可以反复检查GPIO传感器。检查传感器块吗？我猜不是，因为如果它是一个GPIO，它是本地可用的硬件，其结果将立即可用。但我会以两种方式给你答案。

您在此处使用线程的主要内容是重复执行某些操作。如果你想在Twisted中反复做某些事情，那么绝不是使用线程 :)的理由。 Twisted包含一个用于重复性任务的优秀API：LoopingCall。你的榜样，重新写入（再次，假设GPIO调用不会阻塞）使用LoopingCall是这样的：

from somewhere import GPIO 

from twisted.internet import reactor, task 
from autobahn.websocket import WebSocketServerFactory, \ 
           WebSocketServerProtocol, \ 
           listenWS 

class EchoServerProtocol(WebSocketServerProtocol): 

    def check_movement(self): 
     print "checking sensor" 
     sensor_state = GPIO.input(11) 
     if sensor_state == 1: 
      self.send_m("sensor detected movement") 

    def connectionMade(self): 
     WebSocketServerProtocol.connectionMade(self) 
     self.movement_checker = task.LoopingCall(self.check_movement) 
     self.movement_checker.start(4) 

    def onMessage(self, msg, binary): 
     self.movement_checker.stop() 

    def send_m(self, msg): 
     self.sendMessage(msg) 

if __name__ == '__main__': 
    factory = WebSocketServerFactory("ws://localhost:9000") 
    factory.protocol = EchoServerProtocol 
    listenWS(factory) 
    reactor.run() 

当然，有一种情况下，你仍然需要使用线程：如果GPIO检查器（或任何您的循环任务）需要在线程中运行，因为它是库中的一种潜在阻塞操作，无法修改以更好地使用Twisted，并且不希望阻止主循环。

在这种情况下，你仍然要使用LoopingCall，并充分利用其功能又一个的：如果你从函数返回一个Deferred是LoopingCall呼吁，那么它将不会再调用这个函数，直到Deferred火灾。这意味着您可以将任务转移到线程而不用担心主循环堆积对该线程的查询：您可以在线程完成时自动恢复主线程上的循环。

为了让您更具体地了解我的意思，下面是check_movement函数，该函数被修改为与在线程中运行的长时间运行的阻塞调用一起使用，而不是可以在主循环中运行的快速轮询调用：

def check_movement(self): 
    from twisted.internet.threads import deferToThread 
    def get_input(): 
     # this is run in a thread 
     return GPIO.input(11) 
    def check_input(sensor_state): 
     # this is back on the main thread, and can safely call send_m 
     if sensor_state == 1: 
      self.send_m("sensor movement detected") 
    return deferToThread(get_input).addCallback(check_input) 

关于上述示例的其他内容保持完全一样。

Answer 2

在你的例子中有几个因素在起作用。简答题：研究this documentation on threads in Twisted。

• 虽然你不有使用Twisted的反应器使用协议类（线程和协议实现解耦），你叫reactor.run所以所有的下方，我认为适用于你。
• 让Twisted为您创建线程。走出框架可能会让你陷入困境。 IPC的消息传递没有“公共”API（我认为），所以如果你使用Twisted，你几乎需要一路走下去。
• 默认情况下，Twisted不会切换线程来调用您的回调。要从主反应器线程委托给工作者线程（即执行阻塞I/O），您不必自己创建线程，您可以使用reactor.callInThread，它将在工作线程中运行。如果你从不这样做，那么所有东西都在主反应器线程中运行，这意味着例如任何I/O操作都会阻塞反应器线程，并且只有在I/O完成时才能接收任何事件。
• 在工作线程中运行的代码应该使用reactor.callFromThread来做任何不是线程安全的。提供一个回调，它将在主反应堆线程中运行。你在这里比对不起更安全，相信我。
• 以上所有内容也适用于Deferred加工。因此，在设置回调时，不要害怕使用partial(reactor.callFromThread, mycallback)或partial(reactor.callInThread, mycallback)而不是简单的mycallback。我了解到这一点，没有这些，我发现任何阻塞的I/O，我可能做的延期回调要么出错（由于线程安全问题）或阻止主线程。

如果你刚刚开始使用扭曲，这是一个“信任下降”。学会放手管理自己的线索并通过Queue对象等传递信息。一旦你弄清楚Deferred和反应堆是如何工作的（它被称为“Twisted”是有原因的！），这对你来说看起来很自然。 Twisted确实会强制你在函数式编程风格中解耦和分离问题，但是一旦你完成了，我发现它非常干净并且运行良好。

一个建议：我写了一些装饰用我所有的回调函数，这样我没有要不断呼吁callInThread和callFromThread和异常处理整个代码回调设立Deferred;我的装饰器为我启用了这种行为。这可能会防止错误忘记这样做，而且这对于我来说无疑是让Twisted开发更愉快。