在 websocket asyncio 中使进程非阻塞 [英] Making a process non-blocking inside a websocket asyncio

问题描述

我在 python 中使用 websocket 库与前端的 JS 代码进行通信.这是我正在做的事情的总结.当我收到来自客户端的消息时，我会调用 __process_msg，生成一些输出，然后将结果发送回客户端.这里的问题是 __process_msg 很耗时.当我们在 __process_msg 中时，我无法处理来自客户端的任何其他请求.如何使 __process_msg 非阻塞?

I am using websocket library in python to communicate with a JS code in front end. Here is a summary of what I am doing. When I receive a message from the client, I call __process_msg, generate some output, and then send back the result to the client. The problem here is that __process_msg is time-consuming. While we are inside the __process_msg, I cannot serve any other requests from the clients. How can I make the __process_msg non-blocking?

def __start_webtool_server(self, server, ephemeral_port):

    log.info("Starting the Webtool server thread on %s:%s" % (server, ephemeral_port))

    graph_gui_thread = Thread(target=self.__start_webtool_server_run,
                              args=[server, ephemeral_port],
                              daemon=True)
    graph_gui_thread.start()
    return graph_gui_thread


def __start_webtool_server_run(self, server, ephemeral_port):

    try:
        asyncio.set_event_loop(asyncio.new_event_loop())
        start_server = websockets.serve(self.__start_webtool_server_async, server, ephemeral_port, max_size=None)
        asyncio.get_event_loop().run_until_complete(start_server)
        asyncio.get_event_loop().run_forever()
    except Exception as E:
        log.error("Error: %s" % E)


async def __start_webtool_server_async(self, websocket, path):

        async for message in websocket:

            await self.__process_msg(message, websocket)


async def __process_msg(self, message, websocket):

    await self.__process_sub_msg(message, websocket)


async def __process_sub_msg(self, message, websocket):

    ### a long blocking processing here ......

    data_to_send = "blablabla"
    await websocket.send(data_to_send)

推荐答案

如果 __process_msg 正在调用 CPU 绑定或其他非异步(阻塞)代码，您应该将其卸载到线程池以避免在代码运行时停止事件循环.最简单的方法是使用 run_in_executor 事件循环方法:

If __process_msg is invoking CPU-bound or otherwise non-async (blocking) code, you should off-load it to a thread pool to avoid halting the event loop while the code is running. The easiest way to do that is to use the run_in_executor event loop method:

async def __process_msg(self, message, websocket):
    #value = long_running_function(args...)
    loop = asyncio.get_event_loop()
    value = await loop.run_in_executor(None, long_running_function, args...)

这篇关于在 websocket asyncio 中使进程非阻塞的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！