我在业余时间开发维护了一款免费开源的升讯威在线客服系统，也收获了许多用户。对我来说，只要能获得用户的认可，就是我最大的动力。
最近客服系统成功经受住了客户现场组织的压力测试，获得了客户的认可。
客户组织多名客服上线后，所有员工同一时间打开访客页面疯狂不停的给在线客服发消息，系统稳定无异常无掉线，客服回复消息正常。消息实时到达无任何延迟。
我会通过一系列的文章详细分析升讯威在线客服系统的并发高性能技术是如何实现的，使用了哪些方案以及具体的做法。
本篇介绍数据传输方面的管线技术。
Pipelines诞生于.NET Core团队，为使Kestrel成为业界最快的Web服务器之一。最初从作为Kestrel内部的实现细节发展成为可重用的API，它在.Net Core 2.1中作为可用于所有.NET开发人员的最高级BCL API（System.IO.Pipelines）提供。
使用NetworkStream的TCP服务器
在Pipelines之前用.NET编写的典型代码如下所示：
async Task ProcessLinesAsync(NetworkStream stream) { var buffer = new byte[1024]; await stream.ReadAsync(buffer, 0, buffer.Length); // 在buffer中处理一行消息 ProcessLine(buffer); } 此代码可能在本地测试时正确工作，但它有几个潜在错误：
一次ReadAsync调用可能没有收到整个消息（行尾）。
它忽略了stream.ReadAsync()返回值中实际填充到buffer中的数据量。（译者注：即不一定将buffer填充满）
一次ReadAsync调用不能处理多条消息。
这些是读取流数据时常见的一些缺陷。为了解决这个问题，我们需要做一些改变：
我们需要缓冲传入的数据，直到找到新的行。
我们需要解析缓冲区中返回的所有行
async Task ProcessLinesAsync(NetworkStream stream) { var buffer = new byte[1024]; var bytesBuffered = 0; var bytesConsumed = 0; while (true) { var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, buffer.Length - bytesBuffered); if (bytesRead == 0) { // EOF 已经到末尾 break; } // 跟踪已缓冲的字节数 bytesBuffered += bytesRead; var linePosition = -1; do { // 在缓冲数据中查找找一个行末尾 linePosition = Array.IndexOf(buffer, (byte)'n', bytesConsumed, bytesBuffered - bytesConsumed); if (linePosition >= 0) { // 根据偏移量计算一行的长度 var lineLength = linePosition - bytesConsumed; // 处理这一行 ProcessLine(buffer, bytesConsumed, lineLength); // 移动bytesConsumed为了跳过我们已经处理掉的行 (包括n) bytesConsumed += lineLength + 1; } } while (linePosition >= 0); } } 这一次，这可能适用于本地开发，但一行可能大于1KiB（1024字节）。我们需要调整输入缓冲区的大小，直到找到新行。
因此，我们可以在堆上分配缓冲区去处理更长的一行。我们从客户端解析较长的一行时，可以通过使用ArrayPool
async Task ProcessLinesAsync(NetworkStream stream) { byte[] buffer = ArrayPool<byte>.Shared.Rent(1024); var bytesBuffered = 0; var bytesConsumed = 0; while (true) { // 在buffer中计算中剩余的字节数 var bytesRemaining = buffer.Length - bytesBuffered; if (bytesRemaining == 0) { // 将buffer size翻倍 并且将之前缓冲的数据复制到新的缓冲区 var newBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length * 2); Buffer.BlockCopy(buffer, 0, newBuffer, 0, buffer.Length); // 将旧的buffer丢回池中 ArrayPool<byte>.Shared.Return(buffer); buffer = newBuffer; bytesRemaining = buffer.Length - bytesBuffered; } var bytesRead = await stream.ReadAsync(buffer, bytesBuffered, bytesRemaining); if (bytesRead == 0) { // EOF 末尾 break; } // 跟踪已缓冲的字节数 bytesBuffered += bytesRead; do { // 在缓冲数据中查找找一个行末尾 linePosition = Array.IndexOf(buffer, (byte)'n', bytesConsumed, bytesBuffered - bytesConsumed); if (linePosition >= 0) { // 根据偏移量计算一行的长度 var lineLength = linePosition - bytesConsumed; // 处理这一行 ProcessLine(buffer, bytesConsumed, lineLength); // 移动bytesConsumed为了跳过我们已经处理掉的行 (包括n) bytesConsumed += lineLength + 1; } } while (linePosition >= 0); } } 这段代码有效，但现在我们正在重新调整缓冲区大小，从而产生更多缓冲区副本。它将使用更多内存，因为根据代码在处理一行行后不会缩缓冲区的大小。为避免这种情况，我们可以存储缓冲区序列，而不是每次超过1KiB大小时调整大小。
此外，我们不会增长1KiB的 缓冲区，直到它完全为空。这意味着我们最终传递给ReadAsync越来越小的缓冲区，这将导致对操作系统的更多调用。
为了缓解这种情况，我们将在现有缓冲区中剩余少于512个字节时分配一个新缓冲区：
public class BufferSegment { public byte[] Buffer { get; set; } public int Count { get; set; } public int Remaining => Buffer.Length - Count; } async Task ProcessLinesAsync(NetworkStream stream) { const int minimumBufferSize = 512; var segments = new List<BufferSegment>(); var bytesConsumed = 0; var bytesConsumedBufferIndex = 0; var segment = new BufferSegment { Buffer = ArrayPool<byte>.Shared.Rent(1024) }; segments.Add(segment); while (true) { // Calculate the amount of bytes remaining in the buffer if (segment.Remaining < minimumBufferSize) { // Allocate a new segment segment = new BufferSegment { Buffer = ArrayPool<byte>.Shared.Rent(1024) }; segments.Add(segment); } var bytesRead = await stream.ReadAsync(segment.Buffer, segment.Count, segment.Remaining); if (bytesRead == 0) { break; } // Keep track of the amount of buffered bytes segment.Count += bytesRead; while (true) { // Look for a EOL in the list of segments var (segmentIndex, segmentOffset) = IndexOf(segments, (byte)'n', bytesConsumedBufferIndex, bytesConsumed); if (segmentIndex >= 0) { // Process the line ProcessLine(segments, segmentIndex, segmentOffset); bytesConsumedBufferIndex = segmentOffset; bytesConsumed = segmentOffset + 1; } else { break; } } // Drop fully consumed segments from the list so we don't look at them again for (var i = bytesConsumedBufferIndex; i >= 0; --i) { var consumedSegment = segments[i]; // Return all segments unless this is the current segment if (consumedSegment != segment) { ArrayPool<byte>.Shared.Return(consumedSegment.Buffer); segments.RemoveAt(i); } } } } (int segmentIndex, int segmentOffest) IndexOf(List<BufferSegment> segments, byte value, int startBufferIndex, int startSegmentOffset) { var first = true; for (var i = startBufferIndex; i < segments.Count; ++i) { var segment = segments[i]; // Start from the correct offset var offset = first ? startSegmentOffset : 0; var index = Array.IndexOf(segment.Buffer, value, offset, segment.Count - offset); if (index >= 0) { // Return the buffer index and the index within that segment where EOL was found return (i, index); } first = false; } return (-1, -1); } 此代码只是得到很多更加复杂。当我们正在寻找分隔符时，我们同时跟踪已填充的缓冲区序列。为此，我们此处使用List
我们的服务器现在处理部分消息，它使用池化内存来减少总体内存消耗，但我们还需要进行更多更改：
- 我们使用的byte[]和ArrayPool
的只是普通的托管数组。这意味着无论何时我们执行ReadAsync或WriteAsync，这些缓冲区都会在异步操作的生命周期内被固定（以便与操作系统上的本机IO API互操作）。这对GC有性能影响，因为无法移动固定内存，这可能导致堆碎片。根据异步操作挂起的时间长短，池的实现可能需要更改。 - 可以通过解耦读取逻辑和处理逻辑来优化吞吐量。这会创建一个批处理效果，使解析逻辑可以使用更大的缓冲区块，而不是仅在解析单个行后才读取更多数据。这引入了一些额外的复杂性
- 我们需要两个彼此独立运行的循环。一个读取Socket和一个解析缓冲区。
- 当数据可用时，我们需要一种方法来向解析逻辑发出信号。
- 我们需要决定如果循环读取Socket“太快”会发生什么。如果解析逻辑无法跟上，我们需要一种方法来限制读取循环（逻辑）。这通常被称为“流量控制”或“背压”。
- 我们需要确保事情是线程安全的。我们现在在读取循环和解析循环之间共享多个缓冲区，并且这些缓冲区在不同的线程上独立运行。
- 内存管理逻辑现在分布在两个不同的代码段中，从填充缓冲区池的代码是从套接字读取的，而从缓冲区池取数据的代码是解析逻辑。
- 我们需要非常小心在解析逻辑完成之后我们如何处理缓冲区序列。如果我们不小心，我们可能会返回一个仍由Socket读取逻辑写入的缓冲区序列。
复杂性已经到了极端（我们甚至没有涵盖所有案例）。高性能网络应用通常意味着编写非常复杂的代码，以便从系统中获得更高的性能。
System.IO.Pipelines的目标是使这种类型的代码更容易编写。
使用System.IO.Pipelines的TCP服务器
让我们来看看这个例子的样子System.IO.Pipelines：
async Task ProcessLinesAsync(Socket socket) { var pipe = new Pipe(); Task writing = FillPipeAsync(socket, pipe.Writer); Task reading = ReadPipeAsync(pipe.Reader); return Task.WhenAll(reading, writing); } async Task FillPipeAsync(Socket socket, PipeWriter writer) { const int minimumBufferSize = 512; while (true) { // 从PipeWriter至少分配512字节 Memory<byte> memory = writer.GetMemory(minimumBufferSize); try { int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None); if (bytesRead == 0) { break; } // 告诉PipeWriter从套接字读取了多少 writer.Advance(bytesRead); } catch (Exception ex) { LogError(ex); break; } // 标记数据可用，让PipeReader读取 FlushResult result = await writer.FlushAsync(); if (result.IsCompleted) { break; } } // 告诉PipeReader没有更多的数据 writer.Complete(); } async Task ReadPipeAsync(PipeReader reader) { while (true) { ReadResult result = await reader.ReadAsync(); ReadOnlySequence<byte> buffer = result.Buffer; SequencePosition? position = null; do { // 在缓冲数据中查找找一个行末尾 position = buffer.PositionOf((byte)'n'); if (position != null) { // 处理这一行 ProcessLine(buffer.Slice(0, position.Value)); // 跳过 这一行+n (basically position 主要位置？) buffer = buffer.Slice(buffer.GetPosition(1, position.Value)); } } while (position != null); // 告诉PipeReader我们以及处理多少缓冲 reader.AdvanceTo(buffer.Start, buffer.End); // 如果没有更多的数据，停止都去 if (result.IsCompleted) { break; } } // 将PipeReader标记为完成 reader.Complete(); } 我们的行读取器的pipelines版本有2个循环：
FillPipeAsync从Socket读取并写入PipeWriter。
ReadPipeAsync从PipeReader中读取并解析传入的行。
与原始示例不同，在任何地方都没有分配显式缓冲区。这是管道的核心功能之一。所有缓冲区管理都委托给PipeReader/PipeWriter实现。
这使得使用代码更容易专注于业务逻辑而不是复杂的缓冲区管理。
在第一个循环中，我们首先调用PipeWriter.GetMemory(int)从底层编写器获取一些内存; 然后我们调用PipeWriter.Advance(int)告诉PipeWriter我们实际写入缓冲区的数据量。然后我们调用PipeWriter.FlushAsync()来提供数据给PipeReader。
在第二个循环中，我们正在使用PipeWriter最终来自的缓冲区Socket。当调用PipeReader.ReadAsync()返回时，我们得到一个ReadResult包含2条重要信息，包括以ReadOnlySequence
在每个循环结束时，我们完成了reader和writer。这允许底层Pipe释放它分配的所有内存。
System.IO.Pipelines
除了处理内存管理之外，其他核心管道功能还包括能够在Pipe不实际消耗数据的情况下查看数据。
PipeReader有两个核心API ReadAsync和AdvanceTo。ReadAsync获取Pipe数据，AdvanceTo告诉PipeReader不再需要这些缓冲区，以便可以丢弃它们（例如返回到底层缓冲池）。
流量控制
在一个完美的世界中，读取和解析工作是一个团队：读取线程消耗来自网络的数据并将其放入缓冲区，而解析线程负责构建适当的数据结构。通常，解析将比仅从网络复制数据块花费更多时间。结果，读取线程可以轻易地压倒解析线程。结果是读取线程必须减慢或分配更多内存来存储解析线程的数据。为获得最佳性能，在频繁暂停和分配更多内存之间存在平衡。
为了解决这个问题，管道有两个设置来控制数据的流量，PauseWriterThreshold和ResumeWriterThreshold。PauseWriterThreshold决定有多少数据应该在调用PipeWriter.FlushAsync之前进行缓冲停顿。ResumeWriterThreshold控制reader消耗多少后写入可以恢复。
IO调度
通常在使用async / await时，会在线程池线程或当前线程上调用continuation SynchronizationContext。
在执行IO时，对执行IO的位置进行细粒度控制非常重要，这样可以更有效地利用CPU缓存，这对于Web服务器等高性能应用程序至关重要。Pipelines公开了一个PipeScheduler确定异步回调运行位置的方法。这使得调用者可以精确控制用于IO的线程。
实践中的一个示例是在Kestrel Libuv传输中，其中IO回调在专用事件循环线程上运行。
简介
升讯威在线客服与营销系统是一款客服软件，但更重要的是一款营销利器。
- 可以追踪正在访问网站或使用 APP 的所有访客，收集他们的浏览情况，使客服能够主动出击，施展话术，促进成单。
访* 客端在 PC 支持所有新老浏览器。包括不支持 WebSocket 的 IE8 也能正常使用。 - 移动端支持所有手机浏览器、APP、各大平台的公众号对接。
- 支持访客信息互通，可传输访客标识、名称和其它任意信息到客服系统。
- 具备一线专业技术水平，网络中断，拔掉网线，手机飞行模式，不丢消息。同类软件可以按视频方式对比测试。
-
- bilibili 视频：https://www.bilibili.com/video/BV1pK4y1N7UP?t=22
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/1.jpg&w=218&h=124&zc=1)
