WebSocket ACK 机制详解：为什么 TCP 已有 ACK，应用层还要做消息确认、重试与恢复

HZH137940556

444人浏览 · 2026-04-28 22:08:28

HZH137940556 · 2026-04-28 22:08:28 发布

WebSocket ACK 机制详解：为什么 TCP 已有 ACK，应用层还要做消息确认、重试与恢复

mock 了几种ACK 延迟的情况

很多人第一次给 WebSocket 做聊天或指令系统时，都会有一个疑问：

底层已经是 TCP 了，TCP 本身就有 ACK，为什么应用层还要自己再做一层 ACK？

先只抓住一句话：

TCP ACK 解决的是“字节流有没有到达对端 TCP”，应用层 ACK 解决的是“这条业务消息有没有被系统真正确认处理”。

TCP 的可靠性很重要，但它保证的是传输层：

字节流尽量不丢
顺序尽量正确
对端 TCP 栈确认收到

它并不能替业务回答这些问题：

服务端业务逻辑有没有真的处理成功
消息有没有成功落库
接收方页面有没有真正消费到
当前消息该显示“发送中”还是“对方已收”

这也是为什么聊天、通知、控制指令这类业务，经常还要再补一层应用层 ACK。

因为这类消息不是“发出去就行”，而是要知道：

服务端是不是收到了
对方是不是收到了
失败后要不要重试
断线重连后要不要恢复补发

而这些状态，TCP 和 WebSocket 协议本身都不会替你定义。

所以这篇文章接下来不再铺理论，直接结合当前项目这套实现，讲它为什么要加 ACK、怎么拆职责、发送和接收各自会经过哪些方法。

一、当前项目为什么要加应用层 ACK

当前项目里加这层 ACK，解决的是下面几个很实际的问题：

发送后可以拿到明确结果，而不是“发出去就算成功”
可以把消息状态展示成 发送中 / 服务端已收 / 对方已收 / 失败
可以对超时消息做重试
可以把未确认消息落盘，重连后继续补发
可以把 ACK 能力和业务逻辑解耦，不把聊天页写成一团

一句话概括：

这层 ACK 把 WebSocket 从“能发消息”升级成了“可确认、可重试、可恢复的业务消息通道”。

二、当前项目的架构拆分

这套实现不是把所有 ACK 逻辑写进聊天页，而是放在 lib/base/socket/ 和 lib/base/socket/ack/ 这一层做基础能力封装。

这里更适合画依赖关系图，而不是消息流向图。因为在当前项目里：

业务层依赖的是 SocketProtocol 抽象
DI 把 ackSocket 这份 SocketProtocol 绑定到 ACKSocketImpl
ACKSocketImpl 依赖的是 @Named('rawSocket') SocketProtocol
rawSocket 这份协议实例最终才由 WebSocketImpl 提供

核心依赖关系如下：

如果从 injectable 生成的依赖注册角度看，真实绑定关系是：

SocketProtocol(instanceName: 'rawSocket') -> WebSocketImpl
SocketProtocol(instanceName: 'ackSocket') -> ACKSocketImpl
SocketConnectStore -> SocketProtocol(instanceName: 'ackSocket')

所以业务层不是直接引用 ACKSocketImpl，而是先依赖 SocketProtocol，再在运行时拿到被注入的 ackSocket 实例；同理，ACKSocketImpl 内部也不是直接依赖某个裸类型字段名叫 WebSocketImpl，而是依赖 rawSocket 这份 SocketProtocol，只是它当前绑定到的实现类正好是 WebSocketImpl。

如果按职责理解，可以把它拆成三层：

抽象层：SocketProtocol
原始传输层：WebSocketImpl
ACK 增强层：ACKSocketImpl + AckManager + AckLocalStore + AckMessageUtil

三、几个核心类分别负责什么

下面这几个类，是当前项目 ACK 架构的关键。

1. `SocketProtocol`

路径：lib/base/socket/SocketProtocol.dart

它定义的是统一抽象协议：

abstract interface class SocketProtocol {
  bool get isConnected;
  Stream<SocketConnectionEvent> get connectionEvents;
  Stream<SocketData> get messages;
  Future<void> connect(String endpoint);
  Future<SocketAckPayload> send(SocketData message, {Duration? timeout});
  Future<void> disconnect({int? closeCode, String? reason});
  Future<void> dispose();
}

它的作用很明确：

对业务层隐藏具体实现
让业务只依赖“协议能力”，不依赖某个具体类
为 WebSocketImpl 和 ACKSocketImpl 提供统一出口

这层设计是合理的，而且职责非常单一：

只定义能力，不实现策略。

2. `WebSocketImpl`

路径：lib/base/socket/WebSocketImpl.dart

这个类负责最底层的原始 WebSocket 通道能力。

它做的事情主要有：

建立标准 WebSocket 连接
接收底层消息并解析成 SocketData
发送 SocketData
管理连接状态流

看它的 send() 就很容易理解它的边界：

@override
Future<SocketAckPayload> send(SocketData message, {Duration? timeout}) async {
  sendData(message);
  return SocketAckPayload(
    localId: message.localId,
    serverId: '',
    timestamp: DateTime.now().millisecondsSinceEpoch,
    code: 200,
    errorMsg: '',
  );
}

它本质上只是把消息写上连接。

它不负责：

pending 管理
超时控制
重试补发
ACK 识别
本地恢复

所以它的职责也很单一：

只做原始传输层。

3. `ACKSocketImpl`

路径：lib/base/socket/ack/ACKSocketImpl.dart

它是整个 ACK 层的对外入口，也是编排层。

它不是直接去做所有细节，而是把几个组件组合起来：

底层发送交给 WebSocketImpl
pending 生命周期交给 AckManager
ACK 报文识别交给 AckMessageUtil

它的职责主要是：

发送前判断是否要进入 ACK 跟踪
连接前先初始化 pending 恢复
收到 ACK 消息时完成对应等待
收到普通业务消息时自动生成接收 ACK

它的核心发送逻辑很典型：

if (!_incomingMessageHandler.shouldTrackACK(message)) {
  return _sendWithoutAckTracking(message, timeout: timeout);
}

final Future<SocketAckPayload> ackFuture = _pendingQueue.registerPendingMessage(
  message,
  timeout: timeout ?? _defaultAckTimeout,
);

try {
  await _socketProtocol.send(message, timeout: timeout);
} catch (error) {
  await _pendingQueue.failPendingMessage(message.localId, '发送失败: $error');
  rethrow;
}

return ackFuture;

这个类的职责也比较清晰：

它负责编排，不负责底层传输，也不直接负责持久化细节。

4. `AckLocalStore`

路径：lib/base/socket/ack/AckLocalStore.dart

它是 ACK 模块的本地仓库。

负责两类状态：

可持久化的 PendingAckMessage
进程内等待中的 Completer<SocketAckPayload>

最重要的一点是，它把“内存状态”和“写盘动作”收口在一起了。

例如保存 pending 消息时：

Future<void> savePendingMessage(PendingAckMessage message) async {
  _pendingMessages[message.localId] = message;
  await _writeAll(_pendingMessages);
}

这层设计的优点很明显：

外层不用关心什么时候写磁盘
pending 的读写口径统一
有利于后面继续替换存储实现

它的职责也是单一的：

只负责 ACK 本地状态存储，不负责任何协议判断。

5. `AckMessageUtil`

路径：lib/base/socket/ack/AckMessageUtil.dart

这个类很轻，但很重要。

它只做 ACK 报文相关的“判定和组装”：

这条消息是不是 ACK
这条 ACK 能不能解析成 SocketAckPayload
普通业务消息要不要自动回执

例如：

bool isAckMessage(SocketData message) {
  return message.topic == socketAckTopic && message.type == ackType;
}

它的职责是当前架构里最单一的一个：

只做 ACK 报文语义处理。

6. `AckManager`

路径：lib/base/socket/ack/AckManager.dart

这是当前 ACK 实现里最核心的类。

如果一句话概括它的职责，那就是：

负责 pending 消息从“登记”到“确认 / 超时 / 重试 / 恢复”的整个生命周期调度。

它负责的事情包括：

建立 pending 记录
创建 ACK 等待对象
注册超时
连接恢复后补发
定时重试
收到 ACK 后完成 Future
发送失败时清理状态

这个类看起来方法比较多，但职责其实仍然是围绕同一件事：

管理 pending 消息生命周期。

所以它不是“杂”，而是这一块天生就需要它集中收口。

我认为它的单一性是成立的，不过它是这几个类里最“重”的一个。

四、这几个类的职责是否单一

整体上，这套拆分是比较健康的。

可以认为单一性做得比较好的点

SocketProtocol：只负责抽象
WebSocketImpl：只负责原始连接和收发
AckLocalStore：只负责存储
AckMessageUtil：只负责 ACK 报文语义
ACKSocketImpl：只负责协调和编排
AckManager：只负责 pending 生命周期

其中最重的是 `AckManager`

它之所以显得复杂，不是因为职责跑偏，而是因为“pending 生命周期”本身就包含很多状态：

创建
等待
超时
重试
断线
恢复
完成

所以它复杂，但还算集中。

换句话说：

当前这套架构不是“每个类都很小”，而是“每个类的复杂度都尽量待在自己该待的位置”。

这是更重要的事。

五、这套架构的优势是什么

1. 业务层接入很快

业务侧只需要依赖 SocketProtocol，并调用：

connect
send
messages

ACK 是作为基础能力注入进去的，而不是每个业务页面自己手写一套。

2. ACK 和业务解耦

聊天页、RTC 页不需要关心：

pending 落盘
重试时机
ACK 是否超时
收到 ACK 后怎么完成等待

这能明显降低业务层复杂度。

3. 原始通道和增强能力分层

WebSocketImpl 不被 ACK 逻辑污染；
ACKSocketImpl 也不需要重写底层连接能力。

这样后续扩展时会更舒服。

4. 更适合继续演进

未来如果你要继续做这些事：

服务端幂等
更多 ACK stage
失败原因上报
更复杂的恢复策略

这套分层都是比较好接的。

六、ACK 的实现原理

这套 ACK 的实现原理，可以压缩成一句话：

发送时先把消息登记成 pending，再等待 ACK；如果超时还没等到，就按策略重试；如果应用重连，就把 pending 消息恢复出来继续补发。

它依赖几个关键点：

每条消息都有稳定的 localId
ACK 回来时也带同样的 localId
本地持久化 pending 消息
内存里维护等待中的 Completer
定时扫描是否需要重试

七、重点讲 `AckManager` 内部实现

下面只看最核心的几段逻辑。

1. 发送前先登记 pending

Future<SocketAckPayload> registerPendingMessage(
  SocketData message, {
  required Duration timeout,
}) async {
  final PendingAckMessage pendingMessage = _buildPendingMessage(message);
  final String localId = message.localId;
  final Completer<SocketAckPayload> completer = _localStore.createAckCompleter(localId);

  await _localStore.savePendingMessage(pendingMessage);
  _scheduleTimeout(localId: localId, timeout: timeout, completer: completer);
  return completer.future;
}

这里做了三件关键事情：

生成 PendingAckMessage
生成一个等待 ACK 的 Completer
把 pending 写到本地，并注册超时

也就是说，先登记，再发送。

这样即使后面应用闪退，至少 pending 状态已经落盘。

2. 连接成功后触发恢复和补发

void handleSocketConnected() {
  _ensureResendTimer();
  unawaited(_retryPendingMessages(force: true));
}

连接恢复后，它会：

启动重试定时器
立即做一次强制扫描补发

这个 force: true 的意义很重要：

刚重连成功时，不再等下一轮重试间隔，而是优先把遗留消息补出去。

3. 定时扫描哪些消息该重发

Future<void> _retryPendingMessages({bool force = false}) async {
  if (!_canRetryAutomatically || !_isConnected!() || !_localStore.hasPendingMessages) {
    return;
  }

  final DateTime now = DateTime.now();
  final List<PendingAckMessage> nextPendingMessages = <PendingAckMessage>[];

  for (final PendingAckMessage message in _localStore.pendingMessages) {
    if (_hasExceededRetryLimit(message)) {
      _handleRetryExceeded(message.localId);
      continue;
    }
    if (!_shouldResendMessage(message, now, _resendInterval!, force: force)) {
      nextPendingMessages.add(message);
      continue;
    }
    nextPendingMessages.add(_resendMessage(message, now, _sendMessage!));
  }

  await _localStore.replacePendingMessages(nextPendingMessages);
}

这一段是整个 ACK 重试的核心循环。

它会逐条判断：

是否超过最大重试次数
当前是不是到了该重发的时间
如果该重发，就真正调用发送回调
然后把最新状态回写到本地

这里很关键的一点是：

重试不是临时行为，而是一次完整的状态更新。

4. 收到 ACK 后完成等待

void handleReceivedAck(SocketAckPayload ackPayload) {
  final String localId = ackPayload.localId.trim();
  if (localId.isEmpty) {
    return;
  }

  unawaited(_removePendingMessage(localId));
  final Completer<SocketAckPayload>? completer = _localStore.takeAckCompleter(localId);
  if (completer == null || completer.isCompleted) {
    return;
  }

  if (ackPayload.isSuccess) {
    completer.complete(ackPayload);
    return;
  }

  completer.completeError(
    SocketAckException(
      ackPayload.errorMsg.isEmpty ? '发送失败(code: ${ackPayload.code})' : ackPayload.errorMsg,
    ),
  );
}

ACK 回来后，AckManager 会：

用 localId 找到对应 pending
删除本地 pending
取出对应 Completer
成功则 complete
失败则 completeError

这一步完成之后，业务侧 await send() 才真正结束。

5. 超时控制

void _scheduleTimeout({
  required String localId,
  required Duration timeout,
  required Completer<SocketAckPayload> completer,
}) {
  Timer(timeout, () {
    if (completer.isCompleted) {
      return;
    }
    unawaited(_removePendingMessage(localId));
    completer.completeError(SocketAckException('ACK 超时: $localId'));
  });
}

这保证了：

即使服务端一直没有回 ACK，发送方也不会永远卡住。

八、发送一条消息，会经历什么方法

下面按当前项目的真实调用链走一遍。

第一步：业务层调用 `send`

业务层最终会调到：

SocketConnectStore.send(...)
再进入 SocketProtocol.send(...)

而当前注入到业务侧的，是 ACKSocketImpl。

第二步：`ACKSocketImpl.send()`

它会先做三件事：

初始化 pending 恢复是否完成
校验 localId
判断这条消息是否需要 ACK 跟踪

如果不需要 ACK，例如 ACK 报文本身，就直接透传到底层。

如果需要 ACK，就进入下一步。

第三步：`AckManager.registerPendingMessage()`

这里会：

把消息包装成 PendingAckMessage
创建 Completer
保存到 AckLocalStore
注册超时

此时消息已经有了“等待确认”的身份。

第四步：调用底层 `WebSocketImpl.send()`

底层做的事情很简单：

把 SocketData 编码成文本
真正写上 WebSocket

第五步：等待 ACK

只要 ACK 没回来，这条消息就处于 pending 状态。

如果：

收到 ACK：完成 Future
超时未收到：失败
重连后仍未确认：进入恢复补发逻辑

整个发送流程图如下：

九、接收到一条消息，会经历什么

接收链路分成两种情况：

收到的是 ACK 消息
收到的是普通业务消息

第一步：底层 `WebSocketImpl` 收到原始数据

它会把文本解析成 SocketData，然后丢到 messages 流里：

final SocketData? socketMessage = SocketData.tryParseText(rawText);
if (socketMessage == null) {
  // error event
  return;
}
_messageController.add(socketMessage);

第二步：进入 `ACKSocketImpl._handleSocketMessage()`

void _handleSocketMessage(SocketData socketMessage) {
  _handleReceivedAckMessage(socketMessage);
  _messageController.add(socketMessage);
}

这里先过 ACK 逻辑，再把消息继续向业务层分发。

第三步：判断是不是 ACK

_handleReceivedAckMessage() 内部会调用：

AckMessageUtil.tryParseAckPayload()
AckMessageUtil.buildReceiverAckMessage()

情况 A：如果收到的是 ACK

那就会：

解析成 SocketAckPayload
调 AckManager.handleReceivedAck()
完成对应发送等待

情况 B：如果收到的是普通业务消息

那就会：

判断这条消息是否需要自动回执
如果需要，就构造一条 receiver_received ACK
再通过底层异步发回去

也就是说：

接收到普通业务消息时，当前 ACK 层会自动帮你补“接收确认”。

对应流程如下：

十、这套架构怎么样

如果从工程设计角度评价，我认为这套架构整体是合理的，而且适合继续演进。

它最大的优点不是“类多”，而是：

把 ACK 相关的复杂度尽量固定在基础层，而不是扩散到每个业务页面。

我会这样评价它：

SocketProtocol：抽象清楚
WebSocketImpl：边界清楚
ACKSocketImpl：编排职责明确
AckLocalStore：存储收口合理
AckMessageUtil：很轻，职责单一
AckManager：最重，但重在该重的地方

如果后面还要继续增强，我最建议优先补的是：

服务端基于 localId 的幂等去重
更明确的 ACK stage 约定
更细的失败原因

这样这套 ACK 才会从“可用”逐步走向“更可靠”。

参考资料

IETF RFC 6455 - The WebSocket Protocol
https://datatracker.ietf.org/doc/html/rfc6455
IETF RFC 793 - Transmission Control Protocol
https://www.rfc-editor.org/rfc/rfc793
MDN - WebSocket API
https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API
RabbitMQ - Consumer Acknowledgements and Publisher Confirms
https://www.rabbitmq.com/docs/confirms
MQTT.org
https://mqtt.org/

本文关键结论依据：

RFC 6455 只定义 WebSocket 协议本身，不替业务定义消息确认语义
MDN 明确指出标准 WebSocket 接口不支持 backpressure
RabbitMQ 官方文档明确指出 acknowledgements 对 reliable delivery 和 data safety 很重要
MQTT 用 QoS 表达不同级别的可靠消息投递语义

https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

图解 AI Agent：Agent = 模型 + Harness

—指围绕"模型该看到什么"做的系统性设计，是 Harness 工程里最核心的一块，Fowler 那篇文章的原文观点是：上下文工程就是把 guides 和 sensors 交付到 Agent 面前的手段[1]。它标准化的是 Harness 里"工具层"的接口，是 Harness 的一个零件，不是 Harness 本身。Claude Code、Codex、开源的 opencode，内核都是这个循环的工