基于Autobahn+Twisted的WebSocket客户端如何检测sendMessage消息送达状态？

问题背景 #

用户的实际场景与困惑：

我使用Python中的Autobahn库结合Twisted来创建WebSocket服务器与客户端。存在服务器可能宕机，但客户端仍通过sendMessage()持续发送数据包的情况，此时会出现数据包丢失问题。请问是否有方法可以检测数据包是否被服务器正常接收，或者未能送达服务器？

我已实现WebSocketClientProtocol提供的onClose()方法，该方法仅能告知我WebSocket连接已断开，但无法解决我的问题。因为代码中的hello()方法每隔1秒运行一次，无论服务器是否运行都会发送数据包。

代码片段：

# 此方法由WebSocketClientProtocol类提供，连接建立时自动触发
def onOpen(self):
    print("WebSocket connection open.")
def hello():
    b = bytearray([0x11, 0x01, 0x00, 0x01, 0x00])
    self.sendMessage(bytes(b), isBinary=True)
    self.factory.reactor.callLater(1, hello)
# 开始每秒发送消息
hello()

当WebSocket服务器运行时，它应能接收客户端发送的字节数据包。但当服务器宕机时，我希望在调用self.sendMessage(bytes(b), isBinary=True)之前就能知晓服务器的运行状态（运行/停止），从而避免数据包丢失。

解决方案 #

我来给你几个实用的方案，帮你提前判断服务器状态、避免数据包丢失：

1. 维护连接状态标记，发送前检查 #

最直接的方式是在客户端协议类里维护一个连接状态变量，通过onOpen和onClose事件更新它，每次发送前先检查状态：

from autobahn.twisted.websocket import WebSocketClientProtocol, WebSocketClientFactory
from twisted.internet import reactor

class MyClientProtocol(WebSocketClientProtocol):
    def __init__(self):
        super().__init__()
        self.is_connected = False  # 标记当前连接状态

    def onOpen(self):
        print("WebSocket connection open.")
        self.is_connected = True
        self.hello()  # 连接建立后开始发送

    def onClose(self, wasClean, code, reason):
        print(f"WebSocket connection closed: {reason}")
        self.is_connected = False

    def hello(self):
        if self.is_connected:
            b = bytearray([0x11, 0x01, 0x00, 0x01, 0x00])
            # 发送前确认连接正常，同时检查sendMessage返回值
            success = self.sendMessage(bytes(b), isBinary=True)
            if not success:
                print("消息发送失败，可能连接已断开")
                self.is_connected = False
        else:
            print("服务器未连接，跳过本次消息发送")
        # 不管是否发送成功，继续调度下一次发送
        self.factory.reactor.callLater(1, self.hello)

if __name__ == "__main__":
    factory = WebSocketClientFactory("ws://localhost:9000")
    factory.protocol = MyClientProtocol
    reactor.connectTCP("localhost", 9000, factory)
    reactor.run()

2. 启用Ping/Pong心跳机制，确认服务器存活 #

WebSocket原生支持Ping/Pong帧，Autobahn可以自动发送Ping并监听Pong响应，以此判断服务器是否存活。这种方式比单纯的连接状态更可靠，能检测到服务器假死（连接还在但无响应）的情况：

class MyClientProtocol(WebSocketClientProtocol):
    def __init__(self):
        super().__init__()
        self.server_alive = False

    def onOpen(self):
        print("WebSocket connection open.")
        self.server_alive = True
        # 每5秒发送一次Ping
        self.setPingInterval(5)
        # 设置Ping超时：如果10秒内没收到Pong，标记服务器不可用
        self.setPingTimeout(10)
        self.hello()

    def onPong(self, payload):
        # 收到Pong，说明服务器正常
        self.server_alive = True

    def onPingTimeout(self):
        # Ping超时，服务器无响应
        print("Ping超时，服务器可能已宕机")
        self.server_alive = False
        # 主动关闭连接，触发后续重连逻辑
        self.sendClose()

    def onClose(self, wasClean, code, reason):
        print(f"WebSocket connection closed: {reason}")
        self.server_alive = False

    def hello(self):
        if self.server_alive:
            b = bytearray([0x11, 0x01, 0x00, 0x01, 0x00])
            self.sendMessage(bytes(b), isBinary=True)
        else:
            print("服务器无响应，跳过消息发送")
        self.factory.reactor.callLater(1, self.hello)

3. 捕获发送异常，动态调整发送逻辑 #

当连接断开时，调用sendMessage可能会抛出异常（比如ConnectionLost），你可以捕获这些异常，及时标记连接状态并暂停发送：

from twisted.internet.error import ConnectionLost

class MyClientProtocol(WebSocketClientProtocol):
    def __init__(self):
        super().__init__()
        self.is_connected = False

    def onOpen(self):
        print("WebSocket connection open.")
        self.is_connected = True
        self.hello()

    def onClose(self, wasClean, code, reason):
        print(f"WebSocket connection closed: {reason}")
        self.is_connected = False

    def hello(self):
        if self.is_connected:
            b = bytearray([0x11, 0x01, 0x00, 0x01, 0x00])
            try:
                success = self.sendMessage(bytes(b), isBinary=True)
                if not success:
                    raise Exception("消息发送失败")
            except (ConnectionLost, Exception) as e:
                print(f"发送失败: {e}")
                self.is_connected = False
        self.factory.reactor.callLater(1, self.hello)

额外建议 #

• 如果需要确保数据包绝对不丢失，可以实现本地消息队列：当服务器不可用时，把待发送的消息暂存在本地队列，等连接恢复后再批量发送。
• 结合Twisted的重连机制：使用ReconnectingClientFactory代替普通的WebSocketClientFactory，当连接断开时自动尝试重连，重连成功后再恢复发送。

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