Python - 在一个单独的子进程或线程中运行Autobahn | Python asyncio websocket服务器

Question

我有一个基于tkinter的GUI程序在Python 3.4.1中运行.我在程序中运行了几个线程来从各个URL获取JSON数据.我想添加一些WebSocket功能,以便能够允许程序充当服务器,并允许多个客户端通过WebSocket连接到它并交换其他JSON数据.

我正在尝试使用Autobahn | Python WebSocket服务器进行asyncio.

我首先尝试在GUI程序下的单独线程中运行asyncio事件循环.但是,每次尝试都会产生'AssertionError:线程'Thread-1'中没有当前事件循环.

然后,我尝试使用标准库多处理程序包生成一个进程,该程序包在另一个进程中运行asyncio事件循环.当我尝试这个时,我没有得到任何异常,但WebSocket服务器也没有启动.

甚至可以在另一个Python程序的子进程中运行asyncio事件循环吗？

有没有办法将asyncio事件循环集成到当前多线程/ tkinter程序中？

更新 下面是我尝试运行初始测试的实际代码.

from autobahn.asyncio.websocket import WebSocketServerProtocol
from autobahn.asyncio.websocket import WebSocketServerFactory
import asyncio
from multiprocessing import Process

class MyServerProtocol(WebSocketServerProtocol):

   def onConnect(self, request):
      print("Client connecting: {0}".format(request.peer))

   def onOpen(self):
      print("WebSocket connection open.")

   def onMessage(self, payload, isBinary):
      if isBinary:
         print("Binary message received: {0} bytes".format(len(payload)))

      else:
         print("Text message received: {0}".format(payload.decode('utf8')))

      ## echo back message verbatim
      self.sendMessage(payload, isBinary)

   def onClose(self, wasClean, code, reason):
      print("WebSocket connection closed: {0}".format(reason))

def start_server():
   factory = WebSocketServerFactory("ws://10.241.142.27:6900", debug = False)
   factory.protocol = MyServerProtocol
   loop = asyncio.get_event_loop()
   coro = loop.create_server(factory, '10.241.142.27', 6900)
   server = loop.run_until_complete(coro)
   loop.run_forever()
   server.close()
   loop.close()


websocket_server_process = Process(target = start_server)
websocket_server_process.start()

其中大部分直接来自asobio的Autobahn | Python示例代码.如果我尝试将它作为进程运行它没有做任何事情,没有客户端可以连接到它,如果我运行netstat -a没有使用端口6900.如果只在主程序中使用start_server(),则会创建WebSocket服务器.

Answer 1

首先,你得到的AssertionError: There is no current event loop in thread 'Thread-1'.是因为你asyncio需要程序中的每个线程都有自己的事件循环,但它只会在主线程中自动为你创建一个事件循环.因此,如果您asyncio.get_event_loop在主线程中调用一次,它将自动创建一个循环对象并将其设置为默认值,但如果您在子线程中再次调用它,则会收到该错误.相反,您需要在线程启动时显式创建/设置事件循环:

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

完成后,您应该可以get_event_loop()在该特定线程中使用.

可以通过以下方式启动asyncio子进程中的事件循环multiprocessing:

import asyncio
from multiprocessing import Process 

@asyncio.coroutine
def coro():
    print("hi")

def worker():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(coro())

if __name__ == "__main__":
    p = Process(target=worker)
    p.start()
    p.join()

输出:

hi

唯一需要注意的是,如果您在父进程和子进程中启动事件循环,则需要在Unix平台上显式创建/设置新的事件循环(由于Python中的错误)).它应该在Windows上正常工作,或者如果你使用'spawn' multiprocessing上下文.

我认为应该可以asyncio在Tkinter应用程序的后台线程(或进程)中启动事件循环,tkinter并且asyncio并行和事件循环并行运行.如果您尝试从后台线程/进程更新GUI,则只会遇到问题.

Answer 2

@dano的答案可能是正确的，但会创建一个在大多数情况下是不必要的新流程。

我在谷歌上发现了这个问题，因为我自己也有同样的问题。我写了一个应用程序，我希望 websocket api 不在主线程上运行，这导致了您的问题。

我通过简单地阅读 python 文档中的事件循环找到了我的替代解决方案，并找到了解决这个问题的 asyncio.new_event_loop 和 asyncio.set_event_loop 函数。

我没有使用 AutoBahn，而是使用 pypi websockets 库，这是我的解决方案

import websockets
import asyncio
import threading

class WebSocket(threading.Thread):    
    @asyncio.coroutine
    def handler(self, websocket, path):
        name = yield from websocket.recv()
        print("< {}".format(name))
        greeting = "Hello {}!".format(name)
        yield from websocket.send(greeting)
        print("> {}".format(greeting))

    def run(self):
        start_server = websockets.serve(self.handler, '127.0.0.1', 9091)
        eventloop = asyncio.new_event_loop()
        asyncio.set_event_loop(eventloop)
        eventloop.run_until_complete(start_server)
        eventloop.run_forever()

if __name__ == "__main__":
    ws = WebSocket()
    ws.start()