通过一个多小时的代码排查终于查明了线上程序线程数过多的缘由:这是一个接收mq消息的一个服务,程序大致思路是这样的,监听的线程每次收到一条消息,就启动一个线程去执行,每次启动的线程都是新的。说到这里,我们就谈一谈这个程序有哪些弊端呢:前端
线程多的问题该怎么解决呢,增长cpu核心数?治标不治本。对于开发者而言,最为经常使用也最为有效的是线程池化,也就是说线程池。后端
线程池是一种多线程处理形式,处理过程当中将任务添加到队列,而后在建立线程后自动启动这些任务。这避免了在处理短期任务时建立与销毁线程的代价。线程池不只可以保证内核的充分利用,还能防止过度调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。 例如,线程数通常取cpu数量+2比较合适,线程数过多会致使额外的线程切换开销。数组
线程池其中一项很重要的技术点就是任务的队列,队列虽然属于一种基础的数据结构,可是发挥了举足轻重的做用。bash
队列是一种特殊的线性表,特殊之处在于它只容许在表的前端(front)进行删除操做,而在表的后端(rear)进行插入操做,和栈同样,队列是一种操做受限制的线性表。进行插入操做的端称为队尾,进行删除操做的端称为队头。服务器
队列是一种采用的FIFO(first in first out)方式的线性表,也就是常常说的先进先出策略。 网络
public class QueueArray<T>
{
//队列元素的数组容器
T[] container = null;
int IndexHeader, IndexTail;
public QueueArray(int size)
{
container = new T[size];
IndexHeader = 0;
IndexTail = 0;
}
public void Enqueue(T item)
{
//入队的元素放在头指针的指向位置,而后头指针前移
container[IndexHeader] = item;
IndexHeader++;
}
public T Dequeue()
{
//出队:把尾元素指针指向的元素取出并清空(不清空也能够)对应的位置,尾指针前移
T item = container[IndexTail];
container[IndexTail] = default(T);
IndexTail++;
return item;
}
}
复制代码
public class QueueLinkList<T>
{
LinkedList<T> contianer = null;
public QueueLinkList()
{
contianer = new LinkedList<T>();
}
public void Enqueue(T item)
{
//入队的元素其实就是加入到队尾
contianer.AddLast(item);
}
public T Dequeue()
{
//出队:取链表第一个元素,而后把这个元素删除
T item = contianer.First.Value;
contianer.RemoveFirst();
return item;
}
}
复制代码
队列经过数组来实现的话有什么问题吗?是的。首先基于数组不可变本质的因素(具体可参考菜菜以前的文章),当一个队列的元素把数组沾满的时候,数组扩容是有性能问题的,数组的扩容过程不仅是开辟新空间分配内存那么简单,还要有数组元素的copy过程,更可怕的是会给GC形成极大的压力。若是数组比较小可能影响比较小,可是当一个数组比较大的时候,好比占用500M内存的一个数组,数据copy其实会形成比较大的性能损失。数据结构
队列经过数组来实现,随着头指针和尾指针的位置移动,尾指针最终会指向第一个元素的位置,也就是说没有元素能够出队了,其实要解决这个问题有两种方式,其一:在出队或者入队的过程当中不断的移动全部元素的位置,避免上边所说的极端状况发生;其二:能够把数组的首尾元素链接起来,使其成为一个环状,也就是常常说的循环队列。多线程
队列在一些特殊场景下其实还有一些变种,好比说循环队列,阻塞队列,并发队列等,有兴趣的同窗能够去研究一下,这里不在展开讨论。这里说到阻塞队列就多说一句,其实用阻塞队列能够实现一个最基本的生产者消费者模式。并发
当队列用链表方式实现的时候,因为链表的首尾操做时间复杂度都是O(1),并且没有空间大小的限制,因此通常的队列用链表实现更简单。socket
当队列中无元素可出队或者没有空间可入队的时候,是阻塞当前的操做仍是返回错误信息,取决于在座各位队列的设计者了。
//线程池
public class ThreadPool
{
bool PoolEnable = false; //线程池是否可用
List<Thread> ThreadContainer = null; //线程的容器
ConcurrentQueue<ActionData> JobContainer = null; //任务的容器
public ThreadPool(int threadNumber)
{
PoolEnable = true;
ThreadContainer = new List<Thread>(threadNumber);
JobContainer = new ConcurrentQueue<ActionData>();
for (int i = 0; i < threadNumber; i++)
{
var t = new Thread(RunJob);
ThreadContainer.Add(t);
t.Start();
}
}
//向线程池添加一个任务
public void AddTask(Action<object> job,object obj, Action<Exception> errorCallBack=null)
{
if (JobContainer != null)
{
JobContainer.Enqueue(new ActionData { Job = job, Data = obj , ErrorCallBack= errorCallBack });
}
}
//终止线程池
public void FinalPool()
{
PoolEnable = false;
JobContainer = null;
if (ThreadContainer != null)
{
foreach (var t in ThreadContainer)
{
//强制线程退出并很差,会有异常
//t.Abort();
t.Join();
}
ThreadContainer = null;
}
}
private void RunJob()
{
while (true&& JobContainer!=null&& PoolEnable)
{
//任务列表取任务
ActionData job=null;
JobContainer?.TryDequeue(out job);
if (job == null)
{
//若是没有任务则休眠
Thread.Sleep(10);
continue;
}
try
{
//执行任务
job.Job.Invoke(job.Data);
}
catch(Exception error)
{
//异常回调
job?.ErrorCallBack(error);
}
}
}
}
public class ActionData
{
//执行任务的参数
public object Data { get; set; }
//执行的任务
public Action<object> Job { get; set; }
//发生异常时候的回调方法
public Action<Exception> ErrorCallBack { get; set; }
}
复制代码
ThreadPool pool = new ThreadPool(100);
for (int i = 0; i < 5000; i++)
{
pool.AddTask((obj) =>
{
Console.WriteLine($"{obj}__{System.Threading.Thread.CurrentThread.ManagedThreadId}");
}, i, (e) =>
{
Console.WriteLine(e.Message);
});
}
pool.FinalPool();
Console.Read();
复制代码
添加关注,查看更精美版本,收获更多精彩