*【python flask】给多个webocket客户端推送信息_from geventwebsocket.handler import websockethandl

目录

一、初步实现 

服务端

客户端

测试

二、改进

三、异步发送信息

四、调用一次HTTP接口，就发送一次WS信息

一、初步实现 

服务端

 
from flask import Flask,request
from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer
import time,json
 
 
app = Flask(__name__)
 
websocketClient_list = []  # 用于存放所有的websocket客户端对象
@app.route("/")
def index():
    ws = request.environ.get("wsgi.websocket")
    if not ws:  # 使用的是http协议"
        pass
    else:  # 使用的是websocket协议
        websocketClient_list.append(ws)
        while True:
            # 监听连接的websocket客户端，如果message为none，就移除这个客户端
            message = ws.receive()
            if not message:
                websocketClient_list.remove(ws)
                ws.close()
                break
            else:
                # 给所有在线的websocket客户端发信息
                for client in websocketClient_list:
                    client.send(json.dumps({"time:":str(time.time())}, ensure_ascii=False))
                    time.sleep(1)
 
            print("当前在线人数：",len(websocketClient_list))  # 查看当前在线的websocket对象
 
    return "..."
 
if __name__ == '__main__':
    http_server = WSGIServer(("127.0.0.1", 5000), app, handler_class=WebSocketHandler)
    print("server start")
    http_server.serve_forever()

客户端

地址：websocket在线测试

使用该在线工具作为websocket客户端

测试

我这里开了两个客户端：

当服务端收到任何一个客户端发送的消息时，就会给所有在线的客户端发送信息，并打印当前在线人数。当有客户端断开后，就清除这个客户端对象，然后再次打印在线人数。

二、改进

但是上述代码的问题是只有接收到客户端的信息时，服务端才会发送下一条信息。于是我将监听客户端的代码作为一个线程分离出去，就解决了这个问题，如下所示：

（该段代码不论有没有接收到客户端的信息，都会主动向所有在线的客户端发送信息，而且当有客户端断开时，服务端会将这个断开的客户端close）

 
from flask import Flask,request
from geventwebsocket.handler import WebSocketHandler
from gevent.pywsgi import WSGIServer
from gevent import monkey
import time,json,threading
 
 
app = Flask(__name__)
 
monkey.patch_all()
 
def thread_listenClient(ws):
    while True:
        # 监听连接的websocket客户端，如果message为none，就移除这个客户端
        message = ws.receive()
        if not message:
            websocketClient_list.remove(ws)
            ws.close()
            break
 
websocketClient_list = []  # 用于存放所有的websocket客户端对象
@app.route("/")
def index():
    ws = request.environ.get("wsgi.websocket")
 
    thread_obj = threading.Thread(target=thread_listenClient, args=(ws,))
    thread_obj.daemon = True
    thread_obj.start()
 
    if not ws:  # 使用的是http协议"
        pass
 
    else:  # 使用的是websocket协议
        websocketClient_list.append(ws)
        while True:
            # 给所有在线的websocket客户端发信息
            for client in websocketClient_list:
                client.send(json.dumps({"time:":str(time.time())}, ensure_ascii=False))
                time.sleep(1)
 
            print("当前在线人数：",len(websocketClient_list))  # 查看当前在线的websocket对象
 
    return "..."
 
if __name__ == '__main__':
    http_server = WSGIServer(("127.0.0.1", 5000), app, handler_class=WebSocketHandler)
    print("server start")
    http_server.serve_forever()

三、异步发送信息

为了提高消息发送的效率和实时性，我们可以考虑使用异步的方式来发送消息

from flask import Flask, request
import asyncio
import json
import time
import websockets
from threading import Thread
 
app = Flask(__name__)
 
# 存储所有的WS客户端对象
websocketClient_list = []
 
async def send_message():
    while True:
        if websocketClient_list:
            message = json.dumps({"time": str(time.time())}, ensure_ascii=False)
            await asyncio.wait([client.send(message) for client in websocketClient_list])
        await asyncio.sleep(1)
 
async def websocket_handler(websocket, path):
    websocketClient_list.append(websocket)
    try:
        async for message in websocket:
            pass  # 在这里来处理传入的消息
    except websockets.ConnectionClosed:
        pass
    finally:
        websocketClient_list.remove(websocket)
 
@app.route("/framework/pushMessage/framework_czc")
def index():
    return "WebSocket server is running. Connect to ws://<server-address>:8001/websocket"
 
def run_websocket_server():
    asyncio.set_event_loop(asyncio.new_event_loop())
    start_server = websockets.serve(websocket_handler, "0.0.0.0", 8001)
    asyncio.get_event_loop().run_until_complete(start_server)
    asyncio.get_event_loop().run_until_complete(send_message())
    asyncio.get_event_loop().run_forever()
 
if __name__ == '__main__':
    websocket_thread = Thread(target=run_websocket_server)
    websocket_thread.start()
    app.run(host="0.0.0.0", port=8000)

四、调用一次HTTP接口，就发送一次WS信息

from flask import Flask, request, jsonify
import asyncio
import json
import time
import websockets
from threading import Thread
 
app = Flask(__name__)
 
# 存放所有的WS客户端对象
websocketClient_list = []
 
async def send_message(message):
    if websocketClient_list:
        sendData = json.dumps(message, ensure_ascii=False)  # 转成Json格式字符串
        await asyncio.wait([client.send(sendData) for client in websocketClient_list])  # 向所有的WS客户端发送信息
 
async def websocket_handler(websocket, path):
    websocketClient_list.append(websocket)
    try:
        async for message in websocket:
            pass  # 这里可以处理WS传来的信息message
    except websockets.ConnectionClosed:
        pass
    finally:
        websocketClient_list.remove(websocket)
 
@app.route("/framework/pushMessage/framework_czc")  # WS接口
def index():
    return "WebSocket server is running. Connect to ws://<server-address>:8001/websocket"
 
@app.route("/test", methods=['GET','POST'])  # HTTP接口
def connected_clients():
    data = json.loads(request.data)  # 接收Json格式数据
    print(data)
 
    # 发送ws数据
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(send_message(message=data))
    loop.close()
 
    return jsonify({"connected_clients": len(websocketClient_list)})
 
def run_websocket_server():
    asyncio.set_event_loop(asyncio.new_event_loop())
    start_server = websockets.serve(websocket_handler, "0.0.0.0", 8001)
    asyncio.get_event_loop().run_until_complete(start_server)
    asyncio.get_event_loop().run_forever()
 
if __name__ == '__main__':
    websocket_thread = Thread(target=run_websocket_server)
    websocket_thread.start()
    app.run(host="0.0.0.0", port=8000)