从 Python 脚本连接到 websocket 服务器？答案

【问题标题】：Connect to websocket server from Python script?从 Python 脚本连接到 websocket 服务器？
【发布时间】：2019-11-15 06:06:27
【问题描述】：

我有下面的 websocket 服务器从 web 客户端获取连接，并在任何客户端上按下某些键时向所有客户端广播数据。在那个领域一切都很好。但是，我有两个外部 Python 应用程序在服务器上执行各种操作，并且我还需要 websocket 服务器来响应它们的活动。我如何从这些应用程序连接到 websocket 服务器，甚至有可能吗？

websocket 服务器：

class Control(WebSocket):
    def handle(self):
        if self.data == "Left":
            Return = update.leftArrow()
            ClassValues = read_Class()
            LineValue = read_Line()
            message = "Line " + str(ClassValues[Return[1]][LineValue[Return[1]]])
        elif self.data == "Right":
            Return = update.rightArrow()
            LineValue = read_Line()
            message = "Line " + str(ClassValues[Return[1]][LineValue[Return[1]]])
        elif self.data == "Up":
            Return = update.upArrow()
            message = "Control " + str(Return[1])
        elif self.data == "Down":
            Return = update.downArrow()
            message = "Control " + str(Return[1])
        else:
            Return = [False, 0]
        if Return[0]:
            for client in self.server.connections.itervalues():
                client.sendMessage(message)

server = WebSocketServer('', 8000, Control)
server.serve_forever()

【问题讨论】：

标签： python websocket

【解决方案1】：

您可以使用websockets module 连接到您的websocket。我认为该页面上提供的文档比我可以在这里给出的任何解释都要好。它也非常简单易懂。

【讨论】：

【解决方案2】：

根据new-dev-123的输入，我有以下解决方案：

Python

import asyncio
import websockets

if TestCondition:
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(SendMessage(message))

@asyncio.coroutine
def SendMessage(message):
    websocket = yield from websockets.connect('ws://localhost:8000')
    try:
        yield from websocket.send(message)
    finally:
        yield from websocket.close()

Python >= 3.6：

import websockets, asyncio

async def Forward(message):
        url = 'ws://localhost:8000'
        async with websockets.connect(url) as websocket:
                await websocket.send(message)
def xmit_Loop(message):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(Forward(message))

【讨论】：