C#中的高效IO庫System.IO.Pipelines

Posted on by

我們在編寫網絡程序的時候，經常會進行如下操作：

  • 申請一個緩沖區

  • 從數據源中讀入數據至緩沖區

  • 解析緩沖區的數據

  • 重復第2步

表面上看來這是一個很常規而簡單的操作，但實際使用過程中往往存在如下痛點：

數據讀不全：

可能不能在一次read操作中讀入所有需要的數據，因此需要在緩沖區中維護一個遊標，記錄下次讀取操作的起始位置，這個遊標帶瞭瞭不小的復雜度：

  • 從緩沖區讀數據時，要根據遊標計算緩沖區起始寫位置，以及剩餘空間大小。增加瞭讀數據的復雜度。

  • 解析數據也是復用這個緩沖區的，解析的時候也要判斷遊標起始位置，剩餘空間大小。同時增加瞭解析數據的復雜度。

  • 解析玩瞭後還要移動遊標，重新標記緩沖區起始位置，再次增加瞭復雜度。

緩沖區容量有限：

由於緩沖區有限，可能申請的緩沖區不夠用，需要引入動態緩沖區。這也大幅加大瞭代碼的復雜度。

  • 如果每次都申請更大的內存，一方面帶來的內存申請釋放開銷，另一方面需要將原來的數據移動，並更新遊標，帶來更復雜的邏輯。

  • 如果靠多段的內存組成一個邏輯整理，數據的讀寫方式都比較復雜。

  • 使用完後的內存要釋放，如果需要更高的效率還要維持一個內存池。

讀和用沒有分離

我們的業務本身隻關心使用操作，但讀和用操作沒有分離，復雜的都操作導致用操作也變得復雜，並且嚴重幹擾業務邏輯。

今天介紹微軟新推出的一個庫：System.IO.Pipelines（需要在Nuget上安裝），用於解決這些痛點。它主要包含一個Pipe對象，它有一個Writer屬性和Reader屬性。

var pipe   = new Pipe();
var writer = pipe.Writer;
var reader = pipe.Reader;

Writer對象

Writer對象用於從數據源讀取數據，將數據寫入管道中；它對應業務中的"讀"操作。

var content = Encoding.Default.GetBytes("hello world");
var data    = new Memory<byte>(content);
var result  = await writer.WriteAsync(data);

另外，它也有一種使用Pipe申請Memory的方式

var buffer = writer.GetMemory(512);
content.CopyTo(buffer);
writer.Advance(content.Length);
var result = await writer.FlushAsync();

Reader對象

Reader對象用於從管道中獲取數據源，它對應業務中的"用"操作。

首先獲取管道的緩沖區：

var result = await reader.ReadAsync();
var buffer = result.Buffer;

這個Buffer是一個ReadOnlySequence<byte>對象，它是一個相當好的動態內存對象，並且相當高效。它本身由多段Memory<byte>組成，查看Memory段的方法有：

  • IsSingleSegment： 判斷是否隻有一段Memory<byte>

  • First： 獲取第一段Memory<byte>

  • GetEnumerator: 獲取分段的Memory<byte>

它從邏輯上也可以看成一段連續的Memory<byte>，也有類似的方法：

  • Length： 整個數據緩沖區長度

  • Slice： 分割緩沖區

  • CopyTo:　將內容復制到Span中

  • ToArray： 將內容復制到byte[]中

另外，它還有一個類似遊標的位置對象SequencePosition，可以從其Position相關函數中使用，這裡就不多介紹瞭。

這個緩沖區解決瞭"數據讀不夠"的問題，一次讀取的不夠下次可以接著讀，不用緩沖區的動態分配，高效的內存管理方式帶來瞭良好的性能，好用的接口是我們能更關註業務。

獲取到緩沖區後，就是使用緩沖區的數據

var data = buffer.ToArray();

使用完後，告訴PIPE當前使用瞭多少數據，下次接著從結束位置後讀起

reader.AdvanceTo(buffer.GetPosition(4));

這是一個相當實用的設計，它解決瞭"讀瞭就得用"的問題，不僅可以將不用的數據下次再使用，還可以實現Peek的操作，隻讀但不改變遊標。

交互

除瞭"讀"和"用"操作外，它們之間還需要一些交互，例如：

  • 讀過程中數據源不可用，需要停止使用

  • 使用過程中業務結束，需要中止數據源。

Reader和Writer都有一個Complete函數，用於通知結束：

reader.Complete();
writer.Complete();

在Writer寫入和Reader讀取時，會獲得一個結果

FlushResult result = await writer.FlushAsync();
ReadResult result = await reader.ReadAsync();

它們都有一個IsComplete屬性，可以根據它是否為true判斷是否已經結束瞭讀和寫的操作。

取消

在寫入和讀取的時候，也可以傳入一個CancellationToken，用於取消相應的操作。

writer.FlushAsync(CancellationToken.None);
reader.ReadAsync(CancellationToken.None);

如果取消成功，對應的Result的IsCanceled則為true（沒有驗證過）

到此這篇關於C#高效IO庫System.IO.Pipelines的文章就介紹到這瞭。希望對大傢的學習有所幫助，也希望大傢多多支持WalkonNet。

推薦閱讀：