面试题：Python中WebSocket协议的基本应用

import asyncio
import websockets

async def echo(websocket, path):
    # 处理连接建立逻辑
    print('Client connected')
    try:
        async for message in websocket:
            # 接收客户端发送的消息
            print(f'Received: {message}')
            # 将消息回显给客户端
            await websocket.send(message)
            print(f'Sent: {message}')
    except websockets.exceptions.ConnectionClosedOK:
        pass
    # 处理连接关闭逻辑
    print('Client disconnected')

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

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

import asyncio和import websockets：导入asyncio库用于处理异步I/O操作，导入websockets库用于实现WebSocket功能。
async def echo(websocket, path)：定义一个异步函数echo，websocket参数表示与客户端的连接对象，path参数表示连接的路径。
print('Client connected')：当客户端连接时打印提示信息，处理连接建立逻辑。
async for message in websocket:：使用异步循环接收客户端发送的消息。
print(f'Received: {message}')：打印接收到的消息。
await websocket.send(message)：将接收到的消息回显给客户端。
print(f'Sent: {message}')：打印发送的消息。
except websockets.exceptions.ConnectionClosedOK:：捕获连接正常关闭的异常。
print('Client disconnected')：当客户端断开连接时打印提示信息，处理连接关闭逻辑。
start_server = websockets.serve(echo, "localhost", 8765)：创建一个WebSocket服务器，绑定到本地地址localhost的8765端口，echo函数处理连接。
asyncio.get_event_loop().run_until_complete(start_server)：启动事件循环并运行服务器直到完成。
asyncio.get_event_loop().run_forever()：保持事件循环运行，使服务器持续监听连接。

星途面试题库

面试题：Python中WebSocket协议的基本应用

知识考点

面试题答案