您现在的位置：程式碼之家 >> 編程語言 > >> .NET 網頁程式設計 >> 關於.NET

.NET ThreadPool 輪詢調度實現：解決批次任務公平性難題

編輯：關於.NET

在 .NET 應用中，預設 ThreadPool 採用 FIFO 隊列調度，導致先提交的大批次任務佔用所有工作線程，後到達的小批次任務長時間饑餓。本文實現一個 RoundRobinThreadPool，在標準線程池之上構建公平調度層，確保多批次任務獲得均等的執行機會。

核心設計思路

而非單一全局隊列，為每個批次創建獨立隊列。工作線程按輪詢順序依次巡檢各隊列，取出任務執行。這樣：

單批次時，線程專注處理該隊列，效率無損。
多批次時，線程在隊列間輪轉，實現近似公平的時間片分配。
批次完成後，其隊列自動銷毀，剩餘批次自動獲得全部資源。

關鍵數據結構

WorkItem：捕獲上下文的任務封裝

internal class WorkItem {
    public WaitCallback WaitCallback;
    public object State;
    public ExecutionContext Context;

    private static ContextCallback _contextCallback = state => {
        var item = (WorkItem)state;
        item.WaitCallback(item.State);
    };

    public void Execute() {
        if (Context != null)
            ExecutionContext.Run(Context, _contextCallback, this);
        else
            WaitCallback(State);
    }
}

關鍵點：手動捕獲 ExecutionContext（含安全主體、邏輯調用上下文等），並在執行時恢復。這解決了任務調度順序與上下文流動順序不一致的問題，避免安全漏洞。

Queue：批次專屬任務隊列

public sealed class Queue : IDisposable {
    internal Queue<WorkItem> _workItems = new Queue<WorkItem>();
    private RoundRobinThreadPool _pool;
    internal bool _disposed;

    public void QueueUserWorkItem(WaitCallback callback, object state) {
        if (_disposed) throw new ObjectDisposedException(GetType().Name);
        var item = new WorkItem {
            Context = ExecutionContext.Capture(),
            WaitCallback = callback,
            State = state
        };
        lock (_pool._queues) _workItems.Enqueue(item);
        _pool.NotifyNewWorkItem();
    }

    public void Dispose() {
        if (!_disposed) {
            lock (_pool._queues) {
                if (_workItems.Count == 0) _pool.RemoveQueueNeedsLock(this);
            }
            _disposed = true;
        }
    }
}

Dispose 雙重保障：若隊列為空立即移除；否則標記為已棄用，由工作線程在取空後清理。

調度核心：RoundRobinThreadPool

public sealed class RoundRobinThreadPool {
    private List<Queue> _queues;
    private Queue _defaultQueue;
    private int _nextQueue;
    private WaitCallback _callback;

    public RoundRobinThreadPool() {
        _queues = new List<Queue>();
        _callback = DequeueAndExecuteWorkItem;
        _nextQueue = 0;
        _defaultQueue = CreateQueue();
    }

    public Queue CreateQueue() {
        var q = new Queue(this);
        lock (_queues) _queues.Add(q);
        return q;
    }

    private void NotifyNewWorkItem() {
        ThreadPool.UnsafeQueueUserWorkItem(_callback, null);
    }

    private void DequeueAndExecuteWorkItem(object ignored) {
        WorkItem item = null;
        lock (_queues) {
            // 生成輪詢搜索順序：從 _nextQueue 開始循環
            var searchOrder = Enumerable.Range(_nextQueue, _queues.Count - _nextQueue)
                .Concat(Enumerable.Range(0, _nextQueue));

            foreach (int i in searchOrder) {
                var items = _queues[i]._workItems;
                if (items.Count > 0) {
                    item = items.Dequeue();
                    _nextQueue = i;  // 下次從下一個隊列開始
                    if (_queues[i]._disposed && items.Count == 0)
                        RemoveQueueNeedsLock(_queues[i]);
                    break;
                }
            }
            _nextQueue = (_nextQueue + 1) % _queues.Count;
        }
        if (item != null) item.Execute();
    }
    // ... 省略 RemoveQueueNeedsLock 等輔助方法
}

調度邏輯精要

輪詢指針：_nextQueue 記錄上次取任務的隊列，下次從「下一個」開始，避免固定順序導致的飢餓。
原子性操作：單一鎖 _queues 保護隊列列表與項隊列，簡化併發控制（高性能場景可改用細粒度鎖）。
UnsafeQueueUserWorkItem：避免重複流動 ExecutionContext，由 WorkItem 手動管理，提升性能。
自清理機制：工作線程發現已棄用且空的隊列時自動移除，防止記憶體洩漏。

使用示例

private static RoundRobinThreadPool _pool = new RoundRobinThreadPool();

private void ProcessBatch(Batch b) {
    using (var queue = _pool.CreateQueue()) {
        foreach (var item in b) {
            queue.QueueUserWorkItem(() => ProcessItem(item));
        }
    }
}

using 塊確保批次提交完成後隊列被標記棄用，任務執行完畢自動清理。