背压：为慢速 WebSocket 客户端设置发送队列上限 | eSeGeCe 博客

背压：为慢速 WebSocket 客户端设置发送队列上限

· 组件

一台广播高速数据流的服务器，是按自己的节奏产生消息的，而不是按最慢的那个客户端的节奏。行情数据、遥测数据、实时日志追踪：服务器不停地推送，而每一个连上来的客户端都被期待跟得上。它们大多数确实跟得上。但迟早会有一个跟不上。

当一个客户端不再从自己的套接字里取走数据时，你发给它的消息并不会凭空消失。它们会堆积在那个连接的发送队列里，而队列就是内存。一部开进了火车隧道的手机、一个被浏览器限速的标签页、一个不再调用读取的客户端进程，它们全都在悄无声息地推高你服务器的内存占用，直到某个环节崩溃。sgcWebSockets 2026.7 让你可以为那个队列设定上限，并决定它满了之后会发生什么。

慢消费者

问题不在于网络慢，而在于速率不匹配。你的数据流每秒发布 500 条消息。一个链路糟糕的客户端每秒只能取走 50 条。剩下的 450 条总得去某个地方，而这个"某个地方"就是服务器进程里一个按连接维护的列表。一分钟之后，那一个连接就持有着数万条消息。十分钟之后，它持有的就是你的内存了。

损害也不止于那个慢速客户端。它吃掉的内存，就是健康客户端拿不到的内存，而一旦进程开始换页、或者内存分配器开始吃力，所有人的延迟都会跟着遭殃。隧道里的一部手机，就变成了所有人的一次故障。

在此之前，sgcWebSockets 中的发送队列是无界的，这在每个客户端都跟得上的时候没什么问题，但在第一次有客户端跟不上的时候就很难受了。解决办法就是给它加个上限。

给队列设定上限

队列位于组件的 QueueOptions 之下，与已有的 Text、Binary 和 Ping 级别子选项并列。两个新属性为它设定边界：MaxQueueSize 限制单个连接最多可以持有多少条待发消息，而 OverflowPolicy 决定在达到该上限后，下一条消息该如何处理。

uses
  sgcWebSocket, sgcWebSocket_Classes, sgcWebSocket_Types;

begin
  // messages are queued per connection and flushed by the queue thread
  oServer.QueueOptions.Text.Level   := qmLevel1;
  oServer.QueueOptions.Binary.Level := qmLevel1;

  // at most 1000 pending messages per connection
  oServer.QueueOptions.MaxQueueSize := 1000;

  // when the queue is full, throw away the oldest pending message
  oServer.QueueOptions.OverflowPolicy := qopDropOldest;
end;

QueueOptions 在 TsgcWebSocketServer、TsgcWebSocketHTTPServer 和 TsgcWebSocketClient 上都是 published 的，而组件会在接受每个连接时把它复制到该连接上，因此这个上限是针对每个连接的，而不是针对整台服务器的。一千个连接、MaxQueueSize = 1000，最坏情况就是每个连接一千条待发消息，而这个最坏情况现在成了一个你在部署之前就能在纸上乘出来的数字。

选择一种溢出策略

这里没有一个放之四海而皆准的正确答案，所以才有三种策略。你想要哪一种，取决于你的消息意味着什么。

qopDropOldest 适合那种只有最新值才重要的实时数据流。一个价格跳动、一次传感器读数、一个仪表盘指针：下一条消息一到，四秒钟前的那条就一文不值了。从队列头部丢弃意味着，一个落后之后又恢复过来的客户端拿到的是当前的画面，而不是一堆它必须先啃完才能追上进度的陈旧积压。这是默认策略。

qopDropNewest 适合那种较早的消息才重要、而你宁愿卸掉新增负载的场景。如果一个客户端正被引导着走过一个序列，而这个序列的开头才是有意义的部分，那就保留已经排好队的内容，拒绝新来的那一条。

qopDisconnect 适合这种情况：一个跟不上的客户端，就是一个你不想要的客户端。这是最诚实的选项：与其默默地喂给那个连接一份充满缺口、已经损坏的数据流视图，不如直接关掉它。客户端的 WatchDog 会重连，它会重新订阅，并从一个已知的状态重新开始。一个知道自己被断开的客户端可以恢复。一个不知道自己丢了消息的客户端则无法恢复。

// A market data feed: only the newest tick is worth anything.
oServer.QueueOptions.MaxQueueSize   := 500;
oServer.QueueOptions.OverflowPolicy := qopDropOldest;

// A sequence where the head matters: shed the new load instead.
oServer.QueueOptions.OverflowPolicy := qopDropNewest;

// A client that falls this far behind is a client we do not want.
oServer.QueueOptions.OverflowPolicy := qopDisconnect;

不管你选哪一个，都要有意识地选。丢弃消息是一种取舍，而这三个取值的意义就在于，让你自己选择要做哪一种取舍，而不是在生产环境里才发现它。

在丢弃之前先知情：OnSendBufferFull

一个默默丢弃消息的队列，是一个在欺骗你的队列。OnSendBufferFull 就是那个逃生出口。它在任何东西被丢弃之前触发，把连接和当前队列大小交给你，而 var DropMessage 参数让你的代码可以否决这次丢弃。

procedure TForm1.WSServerSendBufferFull(Connection: TsgcWSConnection;
  const QueueSize: Integer; var DropMessage: Boolean);
begin
  // fires BEFORE anything is dropped: log it, count it, alert on it
  DoLog('slow consumer ' + Connection.IP + ' [' + Connection.Guid + '] queue=' +
    IntToStr(QueueSize));
  Inc(FSlowConsumerHits);

  // this connection is a paying feed we must not truncate: keep the message
  // and let the queue grow past the cap for this one send
  if IsPriorityConnection(Connection) then
    DropMessage := False;
end;

设置 DropMessage := False，溢出策略就会对那条消息跳过：它无论如何都会被排入队列，即使超出上限。请节制地使用这一招，因为一个你总是使用的否决，跟根本没有上限是同一回事。这个事件在日常中的用法是那个平淡无奇却又最有价值的用法：知道它发生了、发生在哪个连接上、以及它落后了多远。那个数字就是你的早期预警，远远早于进程耗尽内存的那一刻。

OnSendBufferFull 在 TsgcWebSocketClient、TsgcWebSocketServer 和 TsgcWebSocketHTTPServer 上都是 published 的。

当你根本承受不起丢弃任何消息时该怎么办

有时候，诚实的答案是：一条消息都不能丢。如果你的情况是这样，那么上面三种策略没有一个是你想要的，把队列加大也不是。更大的队列并不能解决问题，它只是把问题推迟了，而且推迟到了一个更糟糕的时刻：更深地陷进一波流量高峰、更深地陷进内存压力，也离你本还能做点什么的那个时间点更远了。

答案是换一种设计。不要把未投递的消息留在套接字队列里，把它们持久化，让客户端在有能力的时候来取回它错过的部分。这正是消息历史与重连恢复所做的事情：服务器为每个通道保留一份带偏移量的有界历史，而一个落后了或者掉线了的客户端会在重新订阅时重放它错过的消息。把它与 qopDisconnect 搭配起来，你就得到了目前最强的组合：一个跟不上的客户端会被迅速切断，而当它回来时，它会恢复所有错过的内容，而不是假装自己没错过。

默认仍然是无上限

MaxQueueSize 默认为 0，意思是无上限，这正是 sgcWebSockets 以前的行为。升级不会改变任何东西。除非你设置一个正数上限，否则溢出路径根本不会被进入，所以现有的服务器会一直按它们一贯的方式排队，直到你决定改变为止。

话虽如此，如果你要向自己无法掌控的客户端广播，而在公共互联网上你永远无法掌控它们，那么一个无界的按连接队列，就是一份代表着"将来会连上来的最糟糕的那个客户端"所做出的、没有上限的内存承诺。设一个上限只需要一行代码。

可用性

发送队列上限、三种溢出策略以及 OnSendBufferFull 随 sgcWebSockets 2026.7 发布，支持 Delphi 7 到 13 以及 C++Builder，平台涵盖 Win32/Win64、Linux64、macOS、Android 和 iOS。

拥有有效订阅的客户可以从客户专区下载新版本。试用用户可以在 esegece.com/products/websockets/download 获取更新后的安装程序。

有疑问、反馈，或者需要帮助为你的数据流挑选合适的策略？联系我们，回复你的将是编写这些代码的人。