================客戶端=================

?

from autobahn.asyncio.websocket import WebSocketClientProtocol, \WebSocketClientFactorytry:import asyncio except ImportError:import trollius as asyncioclass MyClientProtocol(WebSocketClientProtocol):def onConnect(self, response):print("請求連接服務端: {0}".format(response.peer))async def onOpen(self):print("成功服務端連接.")while True:self.sendMessage(u"Hello, world!".encode('utf8'))# self.sendMessage(b"\x00\x01\x03\x04", isBinary=True)await asyncio.sleep(1)def onMessage(self, payload, isBinary):if isBinary:print("接受到服務端的字節信息: {0} bytes".format(len(payload)))else:print("接受到服務端的文本信息: {0}".format(payload.decode('utf8')))def onClose(self, wasClean, code, reason):print("斷開與服務端連接的原因: {0}".format(reason))if __name__ == '__main__':factory = WebSocketClientFactory(u"ws://127.0.0.1:9000")factory.protocol = MyClientProtocolloop = asyncio.get_event_loop()coro = loop.create_connection(factory, '127.0.0.1', 9000)loop.run_until_complete(coro)loop.run_forever()loop.close()

?

=======================服務端========================

from autobahn.asyncio.websocket import WebSocketServerProtocol, \WebSocketServerFactoryclass MyServerProtocol(WebSocketServerProtocol):def onConnect(self, request):print("連接到客戶端: {0}".format(request.peer))def onOpen(self):print("已有客戶端成功連接.")def onMessage(self, payload, isBinary):if isBinary:print("接收到客服端字節信息: {0} bytes".format(len(payload)))else:print("接收到客服端文本信息: {0}".format(payload.decode('utf8')))# echo back message verbatimself.sendMessage(payload, isBinary)def onClose(self, wasClean, code, reason):print("斷開與客戶端的連接原因: {0}".format(reason))if __name__ == '__main__':import asynciofactory = WebSocketServerFactory(u"ws://127.0.0.1:9000")factory.protocol = MyServerProtocolloop = 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:passfinally:server.close()loop.close()

總結

以上是生活随笔為你收集整理的autobahn + Python 简易通信交互的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。