使用WebSockets实现进程间数据交互时遇ConnectionClosedOK问题求助

问题现象 #

尝试通过WebSockets在两个进程间交换数据，首次请求正常响应后，后续请求触发ConnectionClosedOK错误，错误定位在代码行：res = await asyncio.wait_for(websocket.recv(), timeout=10)。

客户端代码 #

import time
import json
import asyncio
import websockets
import nest_asyncio
nest_asyncio.apply()


async def ping(websocket):
    while True:
        await websocket.send('{"message":"PING"}')
        print('------ ping')
        await asyncio.sleep(5)
        
        
async def get_data(websocket, key):
        order = {'key' : key}
        json_data = json.dumps(order)
        await websocket.send(json_data)
        res = await asyncio.wait_for(websocket.recv(), timeout=10)
        res = json.loads(res)
        return res


async def main():
    keeprunning = True
    i = 0
    uri = "ws://localhost:8765"
    async for websocket in websockets.connect(uri, timeout=15, ping_timeout=None, ping_interval=None):
        while keeprunning and i < 10:
            i = i + 1
            print(f'counter {i}')
            res = await get_data(websocket, 'key')
            print(f"res: {res}")
    print('main exit')
    
    
if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

服务端代码 #

import json
import asyncio
import websockets
import nest_asyncio
nest_asyncio.apply()


async def handle_request(websocket):
    message = await websocket.recv()
    data = json.loads(message)
    order = {'key' : data['key'], 'val1' : 1.0, 'val2' : 2.0 }
    json_data = json.dumps(order)
    await websocket.send(json_data)
    print("handle_request")


async def main():
    async with websockets.serve(handle_request, "localhost", 8765, ping_interval=None):
        print('world running...')
        await asyncio.Future()  # run forever
    print('world done')


if __name__ == "__main__":
    asyncio.run(main())

问题原因与修复 #

核心原因 #

服务端handle_request函数仅处理一次客户端请求就返回，函数执行完毕后websockets库会自动关闭当前连接，导致客户端第二次调用websocket.recv()时触发连接已关闭的错误。

修复方案 #

修改服务端handle_request函数，添加循环持续监听客户端消息，直到连接主动关闭：

async def handle_request(websocket):
    try:
        while True:
            message = await websocket.recv()
            data = json.loads(message)
            order = {'key' : data['key'], 'val1' : 1.0, 'val2' : 2.0 }
            json_data = json.dumps(order)
            await websocket.send(json_data)
            print("handle_request")
    except websockets.exceptions.ConnectionClosedOK:
        print("连接已正常关闭")

额外优化建议 #

• 客户端可复用单一连接，避免频繁重建连接：

async def main():
    keeprunning = True
    i = 0
    uri = "ws://localhost:8765"
    async with websockets.connect(uri, timeout=15, ping_timeout=None, ping_interval=None) as websocket:
        while keeprunning and i < 10:
            i = i + 1
            print(f'counter {i}')
            res = await get_data(websocket, 'key')
            print(f"res: {res}")
            await asyncio.sleep(1)  # 避免请求过于频繁
    print('main exit')

内容的提问来源于stack exchange，提问作者Will