怎样用Python开发WebSocket服务？实时通信方案

pip install websockets

import asyncio
import websockets

async def echo(websocket, path):
    async for message in websocket:
        print(f"收到消息: {message}")
        await websocket.send(f"服务器回复: {message}")

start_server = websockets.serve(echo, "localhost", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

async def connect():
    async with websockets.connect("ws://localhost:8765") as websocket:
        await websocket.send("你好")
        response = await websocket.recv()
        print(response)

asyncio.get_event_loop().run_until_complete(connect())

pip install flask-socketio eventlet

from flask import Flask, render_template
from flask_socketio import SocketIO, emit

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)

@socketio.on('connect')
def handle_connect():
    print('客户端已连接')

@socketio.on('message')
def handle_message(data):
    print('收到消息:', data)
    emit('response', f'服务器回应: {data}')

if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=5000)

<script src="https://cdn.socket.io/4.3.2/socket.io.min.js"></script>
<script>
  const socket = io('http://localhost:5000');
  socket.on('connect', () => {
    console.log('已连接到Flask WebSocket');
    socket.emit('message', 'Hello from client');
  });

  socket.on('response', (data) => {
    console.log('收到回复:', data);
  });
</script>

pip install channels

from channels.generic.websocket import AsyncWebsocketConsumer
import json

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']
        await self.send(text_data=json.dumps({'response': message}))

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/chat/$', consumers.ChatConsumer),
]