自製簡易 ThreadPool 實作：突破 .NET 預設限制的多線程方案

在處理高併發任務時，.NET 內建的 ThreadPool 雖然方便，但靈活性有限：無法精確控制線程數量、空閒回收策略固定、任務隊列不可自定義。針對這些痛點，本文分享一個輕量級的自定義線程池 ThreadPoolEx，核心由三大組件構成：ThreadPoolEx（調度核心）、WorkItem（工作單元）、WorkQueue（線程安全隊列）。

核心設計思路

• 動態線程管理：根據待處理任務數與當前線程數自動擴縮容，支援 MinThreadCount 到 MaxThreadCount 範圍。
• 空閒超時回收：線程空閒超過 IdleTimeout 秒且線程數大於最小值時自動退出，避免資源浪費。
• 無鎖計數器：使用 Interlocked 原子操作維護 _inUseWorkThread，減少鎖競爭。

關鍵代碼解析

ThreadPoolEx 主類

public class ThreadPoolEx
{
    private WorkQueue _workQueue = new WorkQueue();
    public int MaxThreadCount = 10;
    public int MinThreadCount = 2;
    private Hashtable _threadTable = null;
    private int _threadCount = 0;
    private int _inUseWorkThread = 0;
    public double IdleTimeout = 10;

    public ThreadPoolEx() : this(10, 2, 2) { }

    public ThreadPoolEx(int maxThreadCount, int minThreadCount, int idleTimeout)
    {
        MaxThreadCount = maxThreadCount;
        MinThreadCount = minThreadCount;
        IdleTimeout = idleTimeout;
        _threadTable = Hashtable.Synchronized(new Hashtable(MaxThreadCount));
    }

    public void QueueUserWorkItem(WaitCallback waitCallback, object objParams)
    {
        EnqueueWorkItem(waitCallback, objParams);
    }

    private void EnqueueWorkItem(WaitCallback waitCallback, object objParams)
    {
        var workItem = new WorkItem
        {
            WorkCallback = waitCallback,
            ObjParams = objParams
        };
        _workQueue.Push(workItem);

        // 需要新線程時啟動
        if (_inUseWorkThread + _waitWorkItem > _threadTable.Count)
        {
            StartThread();
        }
    }
    // ... 省略部分代碼 ...
}

線程工作迴路

private void ProcessWorkItems()
{
    try
    {
        while (true)
        {
            WorkItem workItem = _workQueue.Pop();
            if (workItem == null)
            {
                bool isTimeout = CurThreadIsTimeOut();
                if (isTimeout && _threadTable.Count > MinThreadCount)
                    break; // 退出線程
                Thread.Sleep(100);
            }
            else
            {
                try
                {
                    _threadTable[Thread.CurrentThread] = DateTime.Now;
                    Interlocked.Increment(ref _inUseWorkThread);
                    workItem.Execute();
                }
                catch (Exception)
                {
                    // 記錄日誌
                }
                finally
                {
                    Interlocked.Decrement(ref _inUseWorkThread);
                }
            }
        }
    }
    catch (ThreadAbortException)
    {
        Thread.ResetAbort();
    }
    finally
    {
        if (_threadTable.Count > MinThreadCount)
            _threadTable.Remove(Thread.CurrentThread);
    }
}

WorkQueue 線程安全隊列

class WorkQueue
{
    private static object _lock = new object();
    private Queue<WorkItem> _queue = new Queue<WorkItem>();

    public WorkItem Pop()
    {
        lock (_lock)
        {
            return _queue.Count > 0 ? _queue.Dequeue() : null;
        }
    }

    public void Push(WorkItem item)
    {
        lock (_lock)
        {
            _queue.Enqueue(item);
        }
    }

    public int Count
    {
        get { lock (_lock) return _queue.Count; }
    }
}

WorkItem 工作單元

class WorkItem
{
    public WaitCallback WorkCallback;
    public object ObjParams;

    public void Execute()
    {
        WorkCallback(ObjParams);
    }
}

使用示例

var pool = new ThreadPoolEx(maxThreadCount: 20, minThreadCount: 5, idleTimeout: 30);
for (int i = 0; i < 100; i++)
{
    int taskId = i;
    pool.QueueUserWorkItem(_ =>
    {
        Console.WriteLine($"Task {taskId} running on thread {Thread.CurrentThread.ManagedThreadId}");
        Thread.Sleep(100); // 模擬工作
    }, null);
}

優化方向

1. 引入 BlockingCollection 替代自旋等待，降低 CPU 佔用。
2. 支援任務優先級隊列（PriorityQueue）。
3. 增加全域異常處理與取消令牌（CancellationToken）支援。
4. 實作 IThreadPool 介面，便於單元測試與替換。

完整源碼可在文末下載。歡迎交流改進建議！