前言
我们知道我们的task async 和 await 是基于线程池进行调度的。
但是async 和 await 也就是使用了默认的task调度，让其在线程池中运行。
但是线程池是榨干机器性能为本质的，但是有时候我们运行一些我们自己的需求，比如控制一下线程数（因为并不是线程数越高，就能有更高的性能），控制一下cpu使用，避免cpu使用太高。
正文
首先我们需要一个队列，因为我们需要让task进行保存到某个地方，这里选择队列，因为它简单，也一般符合我们先进先出（先到先运行）的想法。
public sealed class BlockingQueue<T>
{
private readonly Queue<T> _queue = new Queue<T>();
private readonly object _lock = new object();
private readonly Semaphore _pool= new Semaphore(0, int.MaxValue);
public void Enqueue(T item)
{
lock (_lock)
{
_queue.Enqueue(item);
}
}
public T Dequeue()
{
_pool.WaitOne();
lock(_lock)
{
return _queue.Dequeue();
}
}
}
实现一个队列，那么希望是线程安全的，所以要给其进出加上lock。
同时希望,如果队列中为空的时候能够进行等待，不至于一直去轮询。
这里使用的是Semaphore，线程信号量，这个在后面会介绍到。
然后就到了实现线程池调度的时候:
public class TaskThreadPool : TaskScheduler, IDisposable
{
private readonly BlockingQueue<Task> _queue = new BlockingQueue<Task>();
private Thread[] _threads;
private bool _disposed;
private readonly object _lock = new object();
public int ThreadCount { get; }
public TaskThreadPool(int threadCount, bool isBackground = false)
{
if (threadCount < 1)
{
throw new ArgumentOutOfRangeException(nameof(threadCount), "Must be at least 1");
}
ThreadCount = threadCount;
_threads = new Thread[threadCount];
for (int i = 0; i < threadCount; i++)
{
_threads[i] = new Thread(ExcuteTasks)
{
IsBackground = false
};
_threads[i].Start();
}
}
public Task Run(Action action) =>
Task.Factory.StartNew(action, CancellationToken.None, TaskCreationOptions.None, this);
private void ExcuteTasks()
{
while (true)
{
var task = _queue.Dequeue();
if (task == null)
{
return;
}
TryExecuteTask(task);
}
}
protected override IEnumerable<Task>? GetScheduledTasks()
{
return _queue.ToArray();
}
protected override void QueueTask(Task task)
{
_queue.Enqueue(task);
}
protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
{
if (_disposed)
{
throw new ObjectDisposedException(typeof(TaskThreadPool).FullName);
}
return !taskWasPreviouslyQueued && TryExecuteTask(task);
}
public void Dispose()
{
lock (_lock)
{
if (_disposed)
{
return;
}
_disposed = true;
}
for (int i = 0; i < _threads.Length; i++)
_queue.Enqueue(null);
foreach (var thread in _threads)
thread.Join();
_threads = null;
_queue.Dispose();
}
}
代码也很简单常规，就是初始化多少个线程作为线程池，然后Task排队运行就行了,记得要释放资源。
这里dispose让其他线程进行停止的信号为:_queue.Enqueue(null).
private void ExcuteTasks()
{
while (true)
{
var task = _queue.Dequeue();
if (task == null)
{
return;
}
TryExecuteTask(task);
}
}
其他线程消费到null，那么就应该停止了。
然后有一个run方法，可以直接让action，放进来运行:
public Task Run(Action action) =>
Task.Factory.StartNew(action, CancellationToken.None, TaskCreationOptions.None, this);
总结一下基本思路：
需要实现TaskScheduler,这样可以避免自己写一些任务运行的逻辑控制
因为使用了信号量，所以BlockingQueue，然后 TaskThreadPool 需要使用到BlockingQueue,所以需要加上IDispose
需要控制线程数，并在对象销毁的时候禁止新的task进入，运行完已经加入队列的任务
需要有一个run方法，这样对外提供方便
考虑到信号量的释放，那么也完善了blockingqueue:
public sealed class BlockingQueue<T> : IDisposable
{
private readonly Queue<T> _queue = new Queue<T>();
private readonly object _lock = new object();
private readonly Semaphore _pool= new Semaphore(0, int.MaxValue);
public void Enqueue(T item)
{
lock (_lock)
{
_queue.Enqueue(item);
}
}
public T Dequeue()
{
_pool.WaitOne();
lock(_lock)
{
return _queue.Dequeue();
}
}
public IEnumerable<T> ToArray()
{
return _queue.ToArray();
}
public void Dispose()
{
_pool.Dispose();
}
}
结
简单写了一下自定义的线程池，上文中介绍到了Semaphore 这个信号量，下一节为Semaphore 的介绍和实现原理。
