构建高性能C#自定义任务调度器（实现智能负载均衡策略详解）

第一步：创建基础任务调度器类

首先，我们需要继承 TaskScheduler 类，并重写其核心方法：

using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.Threading;using System.Threading.Tasks;public class LoadBalancedTaskScheduler : TaskScheduler{    private readonly List<Thread> _threads;    private readonly ConcurrentQueue<Task> _taskQueue;    private readonly CancellationTokenSource _cts;    public LoadBalancedTaskScheduler(int threadCount = Environment.ProcessorCount)    {        _taskQueue = new ConcurrentQueue<Task>();        _threads = new List<Thread>(threadCount);        _cts = new CancellationTokenSource();        // 启动工作线程        for (int i = 0; i < threadCount; i++)        {            var thread = new Thread(WorkerLoop)            {                IsBackground = true,                Name = $"LoadBalancedWorker-{i}"            };            thread.Start();            _threads.Add(thread);        }    }    protected override void QueueTask(Task task)    {        _taskQueue.Enqueue(task);    }    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)    {        // 简化处理：不支持内联执行        return false;    }    protected override IEnumerable<Task> GetScheduledTasks()    {        return _taskQueue.ToArray();    }    private void WorkerLoop()    {        while (!_cts.Token.IsCancellationRequested)        {            if (_taskQueue.TryDequeue(out Task task))            {                TryExecuteTask(task);            }            else            {                Thread.Sleep(1); // 避免忙等待            }        }    }    public void Dispose()    {        _cts.Cancel();        foreach (var thread in _threads)        {            thread.Join(100); // 等待线程结束        }    }}

第二步：引入负载均衡策略

上面的代码使用了简单的全局队列，所有线程共享同一个任务队列。这在某些情况下会导致竞争激烈。更好的做法是采用“工作窃取”（Work Stealing）策略——每个线程拥有自己的本地队列，当本地无任务时，从其他线程的队列中“窃取”任务。

下面是一个简化版的工作窃取实现：

public class WorkStealingTaskScheduler : TaskScheduler, IDisposable{    private readonly BlockingCollection<Task>[] _localQueues;    private readonly Thread[] _threads;    private readonly CancellationTokenSource _cts;    private readonly Random _random = new Random();    public WorkStealingTaskScheduler(int threadCount = Environment.ProcessorCount)    {        _localQueues = new BlockingCollection<Task>[threadCount];        for (int i = 0; i < threadCount; i++)        {            _localQueues[i] = new BlockingCollection<Task>(new ConcurrentQueue<Task>());        }        _threads = new Thread[threadCount];        _cts = new CancellationTokenSource();        for (int i = 0; i < threadCount; i++)        {            int workerId = i;            _threads[i] = new Thread(() => WorkerLoop(workerId))            {                IsBackground = true,                Name = $"WorkStealingWorker-{workerId}"            };            _threads[i].Start();        }    }    protected override void QueueTask(Task task)    {        // 轮询分配到不同队列，实现初步负载均衡        int targetQueue = _random.Next(_localQueues.Length);        _localQueues[targetQueue].Add(task);    }    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)    {        return false; // 不支持内联    }    protected override IEnumerable<Task> GetScheduledTasks()    {        var allTasks = new List<Task>();        foreach (var queue in _localQueues)        {            allTasks.AddRange(queue);        }        return allTasks;    }    private void WorkerLoop(int workerId)    {        var localQueue = _localQueues[workerId];        while (!_cts.Token.IsCancellationRequested)        {            Task task = null;            // 优先处理本地任务            if (localQueue.TryTake(out task, 10))            {                TryExecuteTask(task);                continue;            }            // 本地无任务，尝试从其他队列窃取            for (int i = 0; i < _localQueues.Length; i++)            {                int stealFrom = (workerId + i + 1) % _localQueues.Length;                if (_localQueues[stealFrom].TryTake(out task))                {                    TryExecuteTask(task);                    break;                }            }            // 如果仍无任务，短暂休眠            if (task == null)            {                Thread.Sleep(1);            }        }    }    public void Dispose()    {        _cts.Cancel();        foreach (var thread in _threads)        {            thread?.Join(100);        }        foreach (var queue in _localQueues)        {            queue?.Dispose();        }    }}

第三步：使用自定义调度器

现在，你可以像这样使用你的新调度器：

using (var scheduler = new WorkStealingTaskScheduler(4)){    var tasks = new List<Task>();    for (int i = 0; i < 100; i++)    {        int taskId = i;        var task = new Task(() =>        {            Console.WriteLine($"Task {taskId} running on thread {Thread.CurrentThread.Name}");            Thread.Sleep(10); // 模拟工作        }, CancellationToken.None, TaskCreationOptions.None, scheduler);        tasks.Add(task);        task.Start(scheduler);    }    Task.WaitAll(tasks.ToArray());}

为什么这属于高级C#并发编程？

通过实现上述调度器，你已经深入掌握了C#并发编程的核心概念：线程管理、无锁队列、任务调度与资源协调。这种能力在构建高性能服务器、游戏引擎或大数据处理平台时至关重要。

总结

本文详细讲解了如何在C#中构建一个支持负载均衡的自定义任务调度器。我们从基础调度器入手，逐步引入工作窃取策略，最终实现了一个高效、可扩展的任务分发系统。掌握这些技术，不仅能提升你的C#任务调度器设计能力，还能让你在面试或实际项目中脱颖而出。

记住，真正的负载均衡算法远比本文示例复杂（例如考虑任务权重、CPU亲和性、I/O等待等），但这个起点足以让你理解核心思想。继续探索吧，未来的高性能系统架构师！