使用 Jetty WebSocket 进行流量控制答案

【问题标题】：Flow control with Jetty WebSocket使用 Jetty WebSocket 进行流量控制
【发布时间】：2013-12-16 08:08:05
【问题描述】：

我正在尝试将 WebSockets 用作应用程序协议（此处没有浏览器）并且遇到了流控制问题...我目前正在使用带有 WebSocketServlet 和 @WebSocket 的 Jetty 9.1.0.RC0。

第一种情况是消息进来的速度比我处理它们的速度快。在这种情况下，我想暂停从该套接字读取并让 TCP 背压来做它的事情。这在以前的应用程序（当然非 websocket）中效果很好。我认为 Session.suspend() 可能会做我想做的事，但它似乎没有任何影响，并且返回的令牌为 null。

有没有办法怀疑阅读？作为一个基于 nio 的连接器，Jetty 将继续读取帧并缓冲它们，直到我可以处理它们为止，我宁愿只是暂停读取而不是引入一些流控制消息。

我的第二种情况是我将 websocket 消息发送到连接的客户端 - 这可能是一个浏览器，但并非总是如此......看来我可以通过使用 sendStringByFuture 方法获得一些流量控制，但有没有办法在未来完成时获得回调，而不是等待或轮询它？

由于我已经嵌入了 jetty，并且正在将它用于 SpringMVC，如果我能在其上获得所需的流控制，以避免在另一个端口上运行类似 Netty 的东西，那就太好了。

谢谢。

【问题讨论】：

Session.suspend() 错误是 Jetty WebSocket 中的一个错误。它将在 9.1 决赛中修复。
是否还会有一个 isReady 等效项来确定写入是否不会阻塞，就像可以为 http 流式传输所做的那样？
Session.suspend() 用于读取暂停，而不是写入。
请注意，WebSocket 可以存在于其他复用并执行它们自己的流控制的协议中（SPDY 中的 WebSocket、HTTP/1.1 + mux-extension 或 HTTP/2 中的 WebSocket），因此暴露基于写的WebSocket API 中的流控制通常被视为短期思维，随着时间的推移变得越来越难以支持。
当我将内容流式传输到我的 websocket 客户端时，在某些时候我需要知道它们是否已备份（在码头中缓冲而不是传输到线路中），所以我的应用程序可以做正确的事。 Jetty 的方式是使用 sendXByFuture，并在下一次发送时检查未来的状态吗？

【解决方案1】：

Jetty 和 JSR API 确实具有异步回调的编写方式。在码头它是： sendBytes(ByteBuffer 数据，WriteCallback 回调) JSR API 是： sendBinary(ByteBuffer 数据，SendHandler 处理程序)

两者都有短信等价物。但这两者都存在一些问题。首先，如果回调用于流量控制，那么回调将“在消息已传输时得到通知”。这对于高吞吐量来说不是最优的，因为这意味着下一条消息的发送将始终等到前一条消息已传输完毕，没有机会将多条消息聚合到一次写入中。

如果后续对sendBytes或sendBinary的调用不等待回调，那么他们可以排队多帧将被高效传输，但问题是发送没有背压，所以内部队列可能永远成长。

一种“解决方案”是限制队列大小，但随后您会遇到对 sendBytes 的所谓异步调用突然阻塞的问题！

我认为解决方案是为发送提供另一种形式的回调，一旦消息排队等待发送，但实际上并未发送。一旦调用了这个回调，那么任何传递的缓冲区都可以被重用/回收，并且可以进行另一个调用和另一个对 sendBytes/sendBinary 的调用。

可能是这样的：

公共接口 QueuedWriteCallback 扩展了 WriteCallback { writeQueued(); }

对于 JSR：

公共接口 QueuedSendHandler 扩展了 SendHandler { onQueued(); }

【讨论】：