ClientWebSocket - MemoryStream、消息帧和性能

Question

这更多的是关于使用 C# ClientWebSocket 时可能的性能优化的理论问题。 考虑以下接收套接字消息的实现。

private async Task ReceiveLoop(CancellationToken cancellationToken, int receiveBufferSize)
{
    try
    {
        WebSocketReceiveResult receiveResult;
        ArraySegment<byte> buffer = new(new byte[receiveBufferSize]);
        do
        {
            await using var stream = new MemoryStream();

            do
            {
                receiveResult = await handler.ReceiveAsync(buffer, cancellationToken);
                await stream.WriteAsync(buffer.AsMemory(0, receiveResult.Count), cancellationToken);

            } while (!receiveResult.EndOfMessage);

            if (receiveResult.MessageType == WebSocketMessageType.Close)
            {
                break;
            }
            var arrayCopy = stream.ToArray();
            // Add to MessageQueue(arrayCopy)
            // Somewhere in the code Consumer, reads the queue and deserializes the array to POCO a few microseconds later
        }
        while (!cancellationToken.IsCancellationRequested);
    }
    catch (OperationCanceledException)
    {
        await DisconnectAsync(cancellationToken);
    }
}

要对此进行一些限制：这里的流的反序列化或编码为 utf8string （本质上是任何副本）似乎是大多数实现的样子。但是，如果流接收大量事件或由于某种原因反序列化失败，则可能会在线程池队列中造成拥塞。 所以当前的实现是添加到队列中并通过单独的Task进行处理。还有围绕不同消息响应类型的逻辑。

如果没有数组副本，流将被释放，反序列化将失败。 另一种方法是投入更多内存来解决问题，例如 ArrayPool.Rent(81920)，并且仅在 Consumer 反序列化后返回数组。如果我们取出 MemoryStream，这会减少 CPU 的负载，但会给分配和 GC 带来压力。 （如果消息 > 81920，也不会扩展消息帧）。

• 有没有一种合理的方法可以在没有副本的情况下重用流？
• 您是否认为在流被处置或数组发生变异之前“复制”到任何类型都是最有意义的？
• 除了使用 MemoryStream 之外还有更好的选择吗？ （RecycableMemoryStream.GetBuffer()？）这可能会提供解决问题的选项？
• 是否还有其他选项，例如保留底层 MemoryStream 缓冲区，直到以安全方式对其进行处理？
• IO Pipelines 在这种情况下会有帮助吗？远离 ClientWebSocket（拥有所有 ping/pong）是最好的选择吗？
• 为了提高该示例的效率，最少可以进行哪些改进？

Answer 1

虽然有一个一般规则是处理所有一次性的东西，但这对于 MemoryStream 来说并不那么重要。它拥有的唯一资源是内存，而内存无论如何都是由垃圾收集器管理的，处置它并没有多大作用。因此，最简单的选择是删除

await using

并将内存流添加到队列中。

另一个选项可能是Microsoft.IO.RecyclableMemoryStream。这个do需要处理流，因为它在内部使用池化缓冲区。这应该会让 GC 对于短暂的流更加友好。

我对 websocket 不太熟悉。但是，如果您可以在开始接收消息时访问消息长度，您应该能够从内存池租用适当大小的缓冲区。这可能会避免大多数分配和复制。

但在做任何事情之前，我建议您进行一些分析和/或基准测试，以确认您确实存在问题。虽然避免分配肯定是一个值得实现的目标，但 GC 相当宽容。