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

Posted 2023-04-18

【中文标题】Python - 在单独的子进程或线程中运行 Autobahn|Python asyncio websocket 服务器

【英文标题】：Python - Running Autobahn|Python asyncio websocket server in a separate subprocess or thread

【发布时间】：2014-09-23 15:56:03

【问题描述】：

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

我正在尝试将 Autobahn|Python WebSocket 服务器用于异步。

我首先尝试在 GUI 程序下的单独线程中运行 asyncio 事件循环。但是，每次尝试都会给出 'AssertionError: There is no current event loop in thread 'Thread-1'。

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

是否可以在另一个 Python 程序的子进程中运行异步事件循环？

有没有办法将 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()

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

【参考方案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()。

可以在通过multiprocessing 启动的子进程中启动asyncio 事件循环：

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 平台上，则需要在子进程中显式创建/设置一个新事件循环（由于bug in Python)。它应该可以在 Windows 上正常工作，或者如果您使用“spawn”multiprocessing 上下文。

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

【讨论】：

我用我试图在单独的进程中运行和使用的确切代码更新了原始帖子。由于某种原因，我使用的代码不起作用。此外，我确实从一些后台线程更新了 GUI。如果我尝试运行 tkinter 和 asyncio，我可能会遇到哪些问题？

@user2662241 据我所知，Tkinter 不支持从主线程以外的任何地方更新 GUI。尝试这样做可能无法正常工作或引发异常。

@user2662241 另外，如果我在上面运行您的确切代码，我最终会得到一个监听端口 6900 的服务器。（不过，这是使用 Python 2.7/trollius 而不是 Python 3.4/asyncio。不立即访问 Python 3.4）。唯一的区别是我使用的是 localhost 而不是您列出的 IP。

谢谢！我在收到此错误消息时遇到了麻烦。我忘了设置事件循环。

【参考方案2】：

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

我在 Google 上找到了这个问题，因为我自己也有同样的问题。我编写了一个应用程序，我希望 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()

【讨论】：

如何通过 websocket “按需”发送消息，即不通过回调？我有一个游戏服务器想通过 websockets 与客户端通信，我想调用 socket_server.sendMessage("my message")，但我只能将它放在回调中......

@andy 您可以将 onConnect() 中的每个新连接添加到存储在类变量中的列表中，然后使用类方法迭代列表以向所有可用连接发送消息（这看起来有点脏——但它对我有用）。不要忘记从 onClose() 中的列表中删除连接。

【参考方案3】：

“有没有办法将异步事件循环集成到当前的多线程/tkinter 程序中？”

是的，使用 asyncio 事件循环运行您的 tkinter 程序。概念证明。

'''Proof of concept integrating asyncio and tk loops.

Terry Jan Reedy
Run with 'python -i' or from IDLE editor to keep tk window alive.
'''

import asyncio
import datetime as dt
import tkinter as tk

loop = asyncio.get_event_loop()
root = tk.Tk()

# Combine 2 event loop examples from BaseEventLoop doc.
# Add button to prove that gui remain responsive between time updates.
# Prints statements are only for testing.

def flipbg(widget, color):
    bg = widget['bg']
    print('click', bg, loop.time())
    widget['bg'] = color if bg == 'white' else 'white'

hello = tk.Label(root)
flipper = tk.Button(root, text='Change hello background', bg='yellow',
                    command=lambda: flipbg(hello, 'red'))
time = tk.Label(root)
hello.pack()
flipper.pack()
time.pack()

def hello_world(loop):
    hello['text'] = 'Hello World'
loop.call_soon(hello_world, loop)

def display_date(end_time, loop):
    print(dt.datetime.now())
    time['text'] = dt.datetime.now()
    if (loop.time() + 1.0) < end_time:
        loop.call_later(1, display_date, end_time, loop)
    else:
        loop.stop()

end_time = loop.time() + 10.1
loop.call_soon(display_date, end_time, loop)

# Replace root.mainloop with these 4 lines.
def tk_update():
    root.update()
    loop.call_soon(tk_update)  # or loop.call_later(delay, tk_update)
# Initialize loop before each run_forever or run_until_complete call    
tk_update() 
loop.run_forever()

我已经用这 4 行额外的行实验性地运行 IDLE，只有在语法突出显示 1000 行时才会注意到减速。