Python+Tornado：WebSocket 心跳包的发送与超时处理

在实时应用中，WebSocket 连接需要保持活跃状态。心跳包机制通过定期发送小数据包检测连接健康度，配合超时处理可自动断开异常连接。以下是完整实现方案：

核心原理

1. 心跳包：服务端定期发送空消息或特定指令
2. 超时检测：记录最后活跃时间，定时检查时差
3. 双定时器：一个用于发送心跳，一个用于检测超时

实现代码

import tornado.ioloop
import tornado.web
import tornado.websocket
import time

class WebSocketHandler(tornado.websocket.WebSocketHandler):
    # 配置参数
    HEARTBEAT_INTERVAL = 15  # 心跳发送间隔(秒)
    TIMEOUT_LIMIT = 40       # 超时断开阈值(秒)
    
    def open(self):
        """连接建立时初始化"""
        self.last_activity = time.time()  # 最后活跃时间戳
        self.heartbeat_timer = None       # 心跳定时器
        self.timeout_timer = None          # 超时检测定时器
        
        # 启动定时器
        self.start_heartbeat()
        self.start_timeout_check()
    
    def on_message(self, message):
        """收到消息时更新活跃时间"""
        self.last_activity = time.time()
    
    def start_heartbeat(self):
        """启动心跳定时器"""
        if self.heartbeat_timer:
            tornado.ioloop.IOLoop.current().remove_timeout(self.heartbeat_timer)
        
        self.heartbeat_timer = tornado.ioloop.IOLoop.current().call_later(
            self.HEARTBEAT_INTERVAL,
            self.send_heartbeat
        )
    
    def send_heartbeat(self):
        """发送心跳包并重置定时器"""
        try:
            if self.ws_connection:  # 检查连接是否存活
                self.write_message("")  # 发送空消息作为心跳
                self.start_heartbeat()  # 重置心跳定时器
        except tornado.websocket.WebSocketClosedError:
            self.close_connection()
    
    def start_timeout_check(self):
        """启动超时检测定时器"""
        if self.timeout_timer:
            tornado.ioloop.IOLoop.current().remove_timeout(self.timeout_timer)
        
        self.timeout_timer = tornado.ioloop.IOLoop.current().call_later(
            self.TIMEOUT_LIMIT,
            self.check_timeout
        )
    
    def check_timeout(self):
        """检测连接超时"""
        idle_time = time.time() - self.last_activity
        if idle_time > self.TIMEOUT_LIMIT:
            print(f"连接超时: {idle_time:.1f}s > {self.TIMEOUT_LIMIT}s")
            self.close_connection()
        else:
            self.start_timeout_check()  # 重置检测定时器
    
    def close_connection(self):
        """安全关闭连接"""
        if self.ws_connection:
            self.close()
        self.cleanup_timers()
    
    def on_close(self):
        """连接关闭时清理资源"""
        self.cleanup_timers()
    
    def cleanup_timers(self):
        """清除所有定时器"""
        if self.heartbeat_timer:
            tornado.ioloop.IOLoop.current().remove_timeout(self.heartbeat_timer)
        if self.timeout_timer:
            tornado.ioloop.IOLoop.current().remove_timeout(self.timeout_timer)
        self.heartbeat_timer = None
        self.timeout_timer = None

# 创建应用
app = tornado.web.Application([
    (r"/ws", WebSocketHandler)
])

if __name__ == "__main__":
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

关键机制说明

1. 双定时器协同：

   • 心跳定时器：每 15 秒触发 send_heartbeat()
   • 超时检测器：每 40 秒触发 check_timeout()

2. 活跃时间更新：

   • 收到任何消息（包括心跳响应）时更新 last_activity
   • 超时检测基于公式：$$ \Delta t = t_{current} - t_{last} $$

3. 异常处理：

   • 发送心跳时捕获 WebSocketClosedError
   • 所有操作前检查 ws_connection 状态

参数优化建议

1. 心跳间隔：根据网络质量调整，公网建议 25-30 秒 $$ T_{heartbeat} \in [15, 30] $$
2. 超时阈值：应大于心跳间隔的 2 倍 $$ T_{timeout} \geq 2 \times T_{heartbeat} $$
3. 重连机制：客户端应实现自动重连逻辑

测试方案

1. 使用浏览器 WebSocket API 连接 ws://localhost:8888/ws
2. 观察控制台输出：
   • 正常连接：每 15 秒输出心跳发送记录
   • 断开网络：40 秒后触发超时断开
3. 网络恢复测试：客户端自动重连

提示：实际部署时应配合 Nginx 反向代理，并配置合适的 proxy_read_timeout 参数，通常设置为 $$ T_{timeout} + 10 $$ 秒。