.NET 流管道並行模式：突破單核限制的加密壓縮加速方案

對大量數據同時進行壓縮與加密時，單線程處理導致雙核 CPU 僅達 50% 佔用。本文實現基於 BlockingStream + 管道並行 的 StreamPipeline，將壓縮與加密分配到不同線程，實測提升 20-25% 吞吐量。

問題根因與優化順序

先壓縮後加密 是鐵律：加密輸出熵值高、不可壓縮。實測《戰爭與和平》文本：先加密後壓縮體積大 250%，耗時增 50%。

管道並行架構

模仿流水線：線程 A 壓縮 → BlockingStream 緩衝 → 線程 B 加密。BlockingStream 為線程安全的生產者-消費者隊列，支援背壓控制。

BlockingStream 核心實現

public class BlockingStream : Stream {
    private Queue<byte[]> _chunks = new Queue<byte[]>();
    private ManualResetEvent _dataAvailable = new ManualResetEvent(false);
    private ManualResetEvent _doneWriting = new ManualResetEvent(false);
    // ... 省略屬性實現

    public override int Read(byte[] buffer, int offset, int count) {
        while (true) {
            int idx = WaitHandle.WaitAny(new[] { _dataAvailable, _doneWriting });
            lock (_lockForRead) lock (_lockForAll) {
                if (_currentChunk == null) {
                    if (_chunks.Count == 0) {
                        if (idx == 1) return 0; // EOF
                        continue;
                    }
                    _currentChunk = _chunks.Dequeue();
                    _currentChunkPosition = 0;
                }
                int avail = _currentChunk.Length - _currentChunkPosition;
                int toCopy = Math.Min(avail, count);
                Buffer.BlockCopy(_currentChunk, _currentChunkPosition, buffer, offset, toCopy);
                _currentChunkPosition += toCopy;
                if (toCopy == avail) {
                    _currentChunk = null;
                    lock (_lockForAll) if (_chunks.Count == 0) _dataAvailable.Reset();
                }
                return toCopy;
            }
        }
    }

    public override void Write(byte[] buffer, int offset, int count) {
        var chunk = new byte[count];
        Buffer.BlockCopy(buffer, offset, chunk, 0, count);
        lock (_lockForAll) {
            if (_illegalToWrite) throw new InvalidOperationException("已關閉");
            _chunks.Enqueue(chunk);
            _dataAvailable.Set();
        }
    }

    public void SetEndOfStream() {
        lock (_lockForAll) { _illegalToWrite = true; _doneWriting.Set(); }
    }
}

StreamPipeline 編排器

public class StreamPipeline : IDisposable {
    private Action<Stream, Stream>[] _filters;
    private List<BlockingStream> _pipes;

    public StreamPipeline(params Action<Stream, Stream>[] filters) {
        _filters = filters;
        _pipes = new List<BlockingStream>(filters.Length - 1);
        for (int i = 0; i < filters.Length - 1; i++) _pipes.Add(new BlockingStream());
    }

    public void Run(Stream input, Stream output) {
        ThreadStart lastStage = null;
        for (int i = 0; i < _filters.Length; i++) {
            Stream inStream = i == 0 ? input : _pipes[i - 1];
            Stream outStream = i == _filters.Length - 1 ? output : _pipes[i];
            var filter = _filters[i];
            ThreadStart stage = () => {
                filter(inStream, outStream);
                if (outStream is BlockingStream bs) bs.SetEndOfStream();
            };
            if (i < _filters.Length - 1) new Thread(stage) { IsBackground = true }.Start();
            else lastStage = stage;
        }
        lastStage(); // 主線程執行最後一階段
    }
    // ... Dispose 實現
}

使用示例

static void Compress(Stream in, Stream out) {
    using (var gz = new GZipStream(out, CompressionMode.Compress, true))
        CopyStream(in, gz);
}

static void Encrypt(Stream in, Stream out) {
    using (var aes = new RijndaelManaged())
    using (var cs = new CryptoStream(out, aes.CreateEncryptor(), CryptoStreamMode.Write))
        CopyStream(in, cs);
}

// 調用
using (var input = File.OpenRead("in.bin"))
using (var output = File.OpenWrite("out.bin"))
using (var pipe = new StreamPipeline(Compress, Encrypt))
    pipe.Run(input, output);

性能分析與擴展建議

• 阿姆達爾定律限制：若壓縮佔 80% 時間，加密佔 20%，理論最大加速 1/(0.8+0.2/2)=1.25倍，實測 1.2-1.25 倍符合預期。
• 避免線程池死鎖：階段間相互等待，使用專用線程而非 ThreadPool。
• 進階優化：結合分塊並行（如並行壓縮多塊再合併），或重用線程池減少創建開銷。