從另一個線程調用一個扭曲的協議方法

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開發更愉快。