Python Autobahn Websocket 服务器一次接收一个连接

Question

Autobahn 使用的 Python Asyncio 或 Twisted 应该同时处理并发连接。 我遵循了关于高速公路阅读文档的一个很好的教程，一切运行良好，但服务器只接收一个连接并处理它的请求，然后接受第二个。

如何确保服务器同时接收多个连接而不占用其他连接对端？

我在网上搜索了一整天，但没有成功 这是我的代码（我在调试时剪掉了很多代码）

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

class NMmapperServerProtocol(WebSocketServerProtocol):
    cmd = NMmapperWSCommandParser() # I have cut out this due to debugging

    def onMessage(self, payload, isBinary):
        """
        @payload the message
        @isBinary whether it's a binary message
        """
        try:
            offload_payload = json.loads(payload.decode("utf-8"))
            await asyncio.gather(cmd.processWSCommands(offload_payload, self))
        except Exception as e:
            raise

    def onConnect(self, request):
        """
        When we've got a peer connect to our server
        """
        try:
            #print(self)
            print(request.peer, "Has connected")
        except Exception as e:
            raise

    def onOpen(self):
        """
        We have a fully connection
        """
        try:
            # Some database action can be made from here
            print("Connection now opened")
        except Exception as e:
            raise

    def onClose(self, wasClean, code, reason):
        """
        @ the client is closing his or her
        connection
        """
        try:
            print("wasClean ", wasClean)
            print("code ", code)
            print("reason ", reason)
        except Exception as e:
            raise

    # Setters
    def setCsrftoken(self, cookie_string):
        """
        @ parse an set
        """
        self.csrftoken = self.parse_csrftoken(cookie_string)

    # Setters
    def setSession(self, cookie_string):
        """
        @ parse an set
        """
        self.session = self.parse_session(cookie_string)


if __name__=="__main__":
    if(IN_PRODUCTION):
        print("RUNNING ")
        factory = NMmapperWSServerFactory(PRODUCTION_HOST, PRODUCTION_PORT)
        factory.run_loop()
    else:
        print("Running on dev")
        factory = WebSocketServerFactory()
        factory.protocol = NMmapperServerProtocol
        
        loop = asyncio.get_event_loop()
        coro = loop.create_server(factory, '0.0.0.0', 9000)
        server = loop.run_until_complete(coro)
        
        try:
            loop.run_forever()
        except KeyboardInterrupt:
            pass
        finally:
            server.close()
            loop.close()

谢谢。

Answer 1

我终于让它按预期工作了。作为一个异步库 我必须在每个执行长时间运行任务的方法上加上 async 前缀

问题出在 onMessage 上，我必须并行处理消息 不要阻止其他想要处理那里的消息的客户端。 所以我必须这样做

offload_payload = json.loads(payload.decode("utf-8"))
loop = asyncio.get_event_loop()

# Offload command processing
loop.create_task(self.processWSCommands(offload_payload, self))

这样每条消息都会被并行处理 即使在这种情况下，也要确保处理消息的方法或函数不会阻塞。

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

class NMmapperServerProtocol(WebSocketServerProtocol):
    cmd = NMmapperWSCommandParser() # I have cut out this due to debugging

    async def onMessage(self, payload, isBinary):
        """
        @payload the message
        @isBinary whether it's a binary message
        """
        try:
            offload_payload = json.loads(payload.decode("utf-8"))
            loop = asyncio.get_event_loop()
            #loop.create_task(runner(10, self.peer))
            #asyncio.gather(runner(20, self.peer))

            # Offload command processing
            loop.create_task(self.processWSCommands(offload_payload, self))
        except Exception as e:
            raise

    def onConnect(self, request):
        """
        When we've got a peer connect to our server
        """
        try:
            #print(self)
            print(request.peer, "Has connected")
        except Exception as e:
            raise

    def onOpen(self):
        """
        We have a fully connection
        """
        try:
            # Some database action can be made from here
            print("Connection now opened")
        except Exception as e:
            raise

    def onClose(self, wasClean, code, reason):
        """
        @ the client is closing his or her
        connection
        """
        try:
            print("wasClean ", wasClean)
            print("code ", code)
            print("reason ", reason)
        except Exception as e:
            raise

    # Setters
    def setCsrftoken(self, cookie_string):
        """
        @ parse an set
        """
        self.csrftoken = self.parse_csrftoken(cookie_string)

    # Setters
    def setSession(self, cookie_string):
        """
        @ parse an set
        """
        self.session = self.parse_session(cookie_string)


if __name__=="__main__":
    if(IN_PRODUCTION):
        print("RUNNING ")
        factory = NMmapperWSServerFactory(PRODUCTION_HOST, PRODUCTION_PORT)
        factory.run_loop()
    else:
        print("Running on dev")
        factory = WebSocketServerFactory()
        factory.protocol = NMmapperServerProtocol

        loop = asyncio.get_event_loop()
        coro = loop.create_server(factory, '0.0.0.0', 9000)
        server = loop.run_until_complete(coro)

        try:
            loop.run_forever()
        except KeyboardInterrupt:
            pass
        finally:
            server.close()
            loop.close()