双WebSocket API连接的心跳处理与频繁断开问题咨询

我来帮你拆解这两个问题，结合WebSocket的常见坑给你分析：

问题1：回复test_message后仍频繁出现1006断开的常见诱因 #

1006是WebSocket的异常关闭码，通常意味着服务器主动判定连接闲置或异常后断开，常见诱因有这些：

• 心跳响应不及时：服务器每10秒发ping，但你的代码可能因为业务逻辑阻塞（比如queue.get()等待时没及时处理消息），错过了回复窗口，服务器判定连接闲置。
• 心跳回复格式不符合要求：很多API要求心跳回复是特定结构（比如JSON格式的{"action": "pong"}），如果你只发了纯字符串test_message，服务器可能没识别这是有效心跳回复，依然判定连接闲置。
• 超时配置不匹配：可能服务器的实际闲置超时不是10秒（比如是2分钟），或者中间的代理/负载均衡（比如Nginx）设置了更短的超时时间，你的心跳频率没覆盖到这个周期。
• 网络层面丢包：偶尔的网络波动会导致ping消息没传到客户端，或者你的回复没传到服务器，服务器触发断开。
• 代码逻辑漏洞：你的两个独立连接中，有一个完全没处理心跳（比如订阅连接），这个连接会因为长期无交互被服务器断开，进而可能引发连锁问题。

问题2：两个websockets.connect连接的属性与心跳处理 #

你代码里的两个websockets.connect(api)是完全独立的WebSocket连接，它们没有共享任何会话属性（比如连接ID、状态），相当于两个不同的客户端同时和服务器建立了连接。

关于心跳处理：
因为服务器会对每个连接单独做闲置检测，所以每个连接都需要单独处理心跳逻辑。你现在的代码里，只有main函数里的连接在尝试处理ping，但subscription_api里的订阅连接完全没处理心跳，这大概率就是问题1里频繁断开的核心原因——这个订阅连接因为没回复心跳被服务器主动断开了！

优化建议：合并连接与心跳逻辑 #

其实完全不需要建立两个连接，把订阅、消息接收、心跳回复整合到同一个连接里更合理，示例代码如下：

import asyncio
import json
import websockets

async def subscription_api(queue, api_url, subscription_msg):
    async with websockets.connect(api_url) as ws:
        # 发送订阅请求
        await ws.send(json.dumps(subscription_msg))
        
        while True:
            try:
                # 接收所有消息（包括ping）
                raw_response = await ws.recv()
                response = json.loads(raw_response)
                
                if response == 'ping':
                    # 回复心跳
                    await ws.send('test_message')
                else:
                    # 把业务消息传递给主函数
                    await queue.put(response)
            except websockets.exceptions.ConnectionClosed:
                # 连接断开时可以尝试重连
                print("连接断开，尝试重连...")
                await asyncio.sleep(5)
                break

async def main():
    queue = asyncio.Queue()
    API_URL = "你的API地址"
    SUBSCRIPTION_MSG = {"type": "subscribe", "channels": ["channel1"]}
    
    # 启动订阅与心跳处理任务
    asyncio.create_task(subscription_api(queue, API_URL, SUBSCRIPTION_MSG))
    
    while True:
        # 从队列获取业务消息并处理
        msg = await queue.get()
        print("收到业务消息:", msg)
        # 这里写你的业务逻辑
        # ...

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

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