Python WebSocket长连接心跳机制与短连接实现详解
本文详细介绍Python中WebSocket长连接与短连接的使用方法。重点解析WebSocketApp类的关键参数，如url、header及各类回调函数。通过示例演示如何配置ping_interval和ping_timeout实现心跳保活，确保连接稳定，并提供短连接的简单创建与关闭代码。
环境安装与长连接概述
安装依赖库
pip install websocket-client首先介绍长连接的基本调用方式：
ws = websocket.WebSocketApp("ws://echo.websocket.org/",
on_message = on_message,
on_error = on_error,
on_close = on_close)
ws.on_open = on_open
ws.run_forever()在建立长连接时，需要理解以下关键参数：
（1）url：指定WebSocket服务器的地址。
（2）header：客户端发送WebSocket握手请求时使用的请求头，格式为字典，例如{'head1':'value1','head2':'value2'}。
（3）on_open：当WebSocket握手成功建立时调用的回调函数。该函数仅接收一个参数，即WebSocketApp类的实例本身。
（4）on_message：当接收到服务器返回的消息时触发。此回调函数包含两个参数：一个是类实例本身，另一个是从服务器接收到的字符串数据（通常为UTF-8编码）。
（5）on_error：当发生错误时调用的回调函数。它接收两个参数：第一个是类实例，第二个是具体的异常对象。
（6）on_close：当连接正常关闭时调用的回调函数。该函数仅接收一个参数，即类实例本身。
（7）on_cont_message：当接收到连续帧数据时触发。该回调函数包含三个参数：类实例、从服务器接收的字符串（UTF-8格式）以及表示是否为连续帧的标志位。
（8）on_data：当从服务器接收到原始消息时调用。该回调函数包含四个参数：类实例、接收到的字符串（UTF-8格式）、数据类型以及连续帧标志。
（9）keep_running：一个布尔型标志位，用于控制应用程序的主循环是否持续运行。默认值为True，若设为False则主循环不会持续运行。
（10）get_mask_key：用于生成WebSocket帧的掩码密钥。
（11）subprotocols：指定一组可用的子协议，默认为空列表。
长连接关键参数与心跳配置
长连接的核心方法是：ws.run_forever(ping_interval=60, ping_timeout=5)
如果不主动关闭WebSocket连接，程序将一直阻塞在此处。该函数支持传入两个参数，若提供这些参数，则会自动启动心跳包发送机制。
ping_interval：设置自动发送“ping”命令的时间间隔（单位：秒）。如果设置为0，则不会自动发送心跳包。
ping_timeout：设置等待“pong”响应的时间（单位：秒）。如果在此时间内未收到pong消息，则视为超时。
ws.run_forever(ping_interval=60,ping_timeout=5)#ping_interval心跳发送间隔时间#ping_timeout 设置，发送ping到收到pong的超时时间查看源代码可以发现如下逻辑片段：
注意，ping的超时时间必须大于ping间隔时间，否则将抛出异常。
if not ping_timeout or ping_timeout <= 0:
ping_timeout = None
if ping_timeout and ping_interval and ping_interval <= ping_timeout:
raise WebSocketException("Ensure ping_interval > ping_timeout")长连接代码示例
以下是长连接的具体实现示例：
示例1：基础长连接实现
import websocket
try:
import thread
except ImportError:
import _thread as thread
import time
def on_message(ws, message):
print(message)
def on_error(ws, error):
print(error)
def on_close(ws):
print("### closed ###")
def on_open(ws):
def run(*args):
ws.send("hello1")
time.sleep(1)
ws.close()
thread.start_new_thread(run,())
if __name__ == "__main__":
websocket.enableTrace(True)
ws = websocket.WebSocketApp("ws://echo.websocket.org/",
on_message = on_message,
on_error = on_error,
on_close = on_close)
ws.on_open = on_open
ws.run_forever(ping_interval=60,ping_timeout=5)示例2：基于线程的长连接实现
import websocket
from threading import Thread
import time
import sys
class MyApp(websocket.WebSocketApp):
def on_message(self, message):
print(message)
def on_error(self, error):
print(error)
def on_close(self):
print("### closed ###")
def on_open(self):
def run(*args):
for i in range(3):
# send the message, then wait
# so thread doesn't exit and socket
# isn't closed
self.send("Hello %d" % i)
time.sleep(1)
time.sleep(1)
self.close()
print("Thread terminating...")
Thread(target=run).start()
if __name__ == "__main__":
websocket.enableTrace(True)
if len(sys.argv) < 2:
host = "ws://echo.websocket.org/"
else:
host = sys.argv[1]
ws = MyApp(host)
ws.run_forever()短连接代码示例
以下是短连接的实现方式：
from websocket import create_connection
ws = create_connection("ws://echo.websocket.org/")
print("Sending 'Hello, World'...")
ws.send("Hello, World")
print("Sent")
print("Receiving...")
result = ws.recv()
print("Received '%s'" % result)
ws.close()以上即为Python WebSocket长连接心跳机制与短连接的完整示例，更多关于Python WebSocket连接的资料请关注本站其它相关文章！
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
