使用Tornado web框架執行異步任務

Question

我熟悉事件驅動的編程，但是我遇到了這個問題，我終止了可能的解決方案。我讀了龍捲風的文檔，我試着用：

1. 期貨
2. gen.coroutine
3. 異步
4. add_timeout

，但我沒能解決以下問題：

• 我有一個websocket服務器這只是聽新郵件 並根據消息類型調用特定的功能

  類WebSocketHandler（tornado.websocket.WebSocketHandler）：

  ... 
  
  def on_message(self, message): 
      if message['type'] is X: 
       self.write(functionA(message['data'])) 
      elif message['type'] is Y: 
       self.write(functionB(message['data'])) 
  ... 
  

問題是當一個計算代價函數是執行，讓說功能A，它可能需要長達5分鐘終止

def functionA(message): 
    params = extract_params(message) 
    cmd = "computationally_expensive_tool" 
    out = check_output(cmd, shell=True, stderr=STDOUT, cwd=working_dir) 
    ... 
    return json.dumps({ 
         "error": False, 
         "message": "computationally_expensive_tool_execution_terminated", 
         "type": X 
        }) 

我的問題是我該如何執行該函數n以異步的方式處理，這樣我仍然可以處理其他消息和functionA的結果。

Answer 1

如果functionA是不能進行異步阻塞函數，你可能想在一個線程池運行：

executor = concurrent.futures.ThreadPoolExecutor() 

@gen.coroutine 
def on_message(self, message): 
    if message['type'] is X: 
     yield executor.submit(functionA, message['data']) 
    elif message['type'] is Y: 
     functionB(message['data']) 

這將阻止該網頁套接字直到functionA結束，但允許其他連接繼續工作。如果需要繼續處理來自相同連接的其他類型的消息，而functionA則運行，則需要更復雜的安排，可能涉及tornado.queues.Queue。