大量數據轉錄的多線程同步實現：ReaderWriterLockSlim + Queue<T> 生產消費模式

背景與挑戰

專案需將兩種異構存儲設備的海量數據進行格式轉換並遷移。單線程順序讀寫耗時過長，必須引入多線程並行。核心難點在於：

• 讀取端（外部設備）與寫入端（本地設備）速率不匹配，需緩衝隊列解耦；
• 多線程併發訪問共享隊列，必須保證線程安全且鎖開銷最小；
• 數據對象在隊列中移動後即被丟棄，無法在隊列層面做唯一性校驗，需由下游存儲層保證。

架構設計

+----------------+      +----------------------+      +------------------+
|  讀取線程 (1)  | ---> |  SynchronizedQueue   | ---> |  寫入線程 (N)    |
|  ReadFromExt() |      |  ReaderWriterLockSlim|      |  SaveToLocal()   |
+----------------+      +----------------------+      +------------------+

• 單生產者：一條線程負責從外部設備讀取，序列化後入隊；
• 多消費者：多條線程從隊列取出，格式轉換後寫入本地設備；
• 同步隊列：基於 Queue<T> 封裝 ReaderWriterLockSlim，提供讀寫分離鎖，支援超時嘗試寫入，防止生產者無限阻塞。

核心程式碼實現

1. 統一讀寫介面

namespace Common {
    public interface IReaderWriter<T> {
        T Read(int argument);
        void Write(int argument, T instance);
        void Delete(int argument);
        void Clear();
    }
}

2. 隊列實現

public class QueueReaderWriter<T> : IReaderWriter<T> {
    private readonly Queue<T> _queues;
    public QueueReaderWriter() : this(0) { }
    public QueueReaderWriter(int capacity) { _queues = new Queue<T>(capacity); }

    public T Read(int argument) => _queues.FirstOrDefault();
    public void Write(int argument, T instance) => _queues.Enqueue(instance);
    public void Delete(int argument) => _queues.Dequeue();
    public void Clear() { _queues.Clear(); _queues.TrimExcess(); }
}

3. 線程安全包裝器

public class SynchronizedWriteData<T> : IDisposable {
    private readonly ReaderWriterLockSlim _lock = new ReaderWriterLockSlim();
    private readonly IReaderWriter<T> _inner;

    public SynchronizedWriteData(IReaderWriter<T> inner) { _inner = inner; }

    // 讀取（共享鎖）
    public T Read() {
        _lock.EnterReadLock();
        try { return _inner.Read(0); }
        finally { _lock.ExitReadLock(); }
    }

    // 寫入（獨佔鎖）
    public void Add(T instance) {
        _lock.EnterWriteLock();
        try { _inner.Write(0, instance); }
        finally { _lock.ExitWriteLock(); }
    }

    // 帶超時寫入，防止生產者無限等待
    public bool AddWithTimeout(T instance, int timeoutMs) {
        if (_lock.TryEnterWriteLock(timeoutMs)) {
            try { _inner.Write(0, instance); return true; }
            finally { _lock.ExitWriteLock(); }
        }
        return false; // 超時可選擇丟棄、重試或記錄日誌
    }

    public void Delete() {
        _lock.EnterWriteLock();
        try { _inner.Delete(0); }
        finally { _lock.ExitWriteLock(); }
    }

    public void Dispose() {
        _lock.EnterWriteLock();
        try { _inner.Clear(); }
        finally { _lock.ExitWriteLock(); _lock.Dispose(); }
    }
}

4. 外部數據適配器抽象基類

public abstract class ExternalDataAdapter<T> : IDisposable {
    protected abstract string ConnectString { get; }
    protected abstract void Initialize();
    public abstract void Transmit();
    protected abstract void ReadFromExternalDevice();
    protected abstract void SaveToInternalDevice();
    public abstract void Dispose();
}

5. 多線程協調類

public abstract class MultiThreadAdapter<T> : ExternalDataAdapter<T> {
    protected SynchronizedWriteData<T> _data;
    protected Thread _readThread;

    public sealed override void Transmit() {
        _readThread = new Thread(ReadFromExternalDevice) { IsBackground = true };
        _readThread.Start();
        Thread.Sleep(10000); // 給生產者預填充時間，避免消費者空跑

        while (_readThread.IsAlive) {
            SaveToInternalDevice(); // 消費者循環取數據
        }
        _readThread.Join(); // 等待生產者結束
    }

    protected abstract override void ReadFromExternalDevice();
    protected abstract override void SaveToInternalDevice();

    public override void Dispose() { _data?.Dispose(); }
}

關鍵設計決策解析

擴展建議

• 背壓控制：在 SynchronizedWriteData 中加入最大容量限制，達到上限時 AddWithTimeout 返回 false，生產者可選擇暫停讀取或丟棄非關鍵數據。
• 批次寫入：消費端 SaveToInternalDevice 可一次 Dequeue 多筆（如 while(count-- > 0 && queue.TryDequeue(out var item))），配合資料庫批次插入（SqlBulkCopy、INSERT ... VALUES (...),(...)）大幅提升寫入吞吐。
• 監控指標：暴露隊列當前長度、生產/消費速率、錯誤計數，接入 Prometheus/Grafana 做實時觀測。
• 異常重試與死信隊列：寫入失敗時進入重試隊列，超過次數轉入死信表人工介入。

總結

基於 ReaderWriterLockSlim 封裝 Queue<T> 實現的生產消費模式，以極低的鎖爭用實現了高吞吐的異構數據遷移。抽象出 IReaderWriter<T>、ExternalDataAdapter<T> 等介面，使具體設備驅動開發聚焦於協議解析與格式轉換，複用同步調度邏輯。實踐中配合批次寫入、背壓控制、監控告警，可平穩支撐 TB 級數據遷移任務。