浅谈.NET下的多线程和并行计算(六)线程池基础下

简介: 这节我们按照线程池的核心思想来自定义一个简单的线程池: 1) 池中使用的线程不少于一定数量,不多于一定数量 2) 池中线程不够的时候创建,富裕的时候收回 3) 任务排队,没有可用线程时,任务等待 我们的目的只是实现这些“需求”,不去考虑性能(比如等待一段时间再去创建新的线程等策略)以及特殊的处理(异常),在实现这个需求的过程中我们也回顾了线程以及线程同步的基本概念。

这节我们按照线程池的核心思想来自定义一个简单的线程池:

1) 池中使用的线程不少于一定数量,不多于一定数量

2) 池中线程不够的时候创建,富裕的时候收回

3) 任务排队,没有可用线程时,任务等待

我们的目的只是实现这些“需求”,不去考虑性能(比如等待一段时间再去创建新的线程等策略)以及特殊的处理(异常),在实现这个需求的过程中我们也回顾了线程以及线程同步的基本概念。

首先,把任务委托和任务需要的状态数据封装一个对象:

public class WorkItem
{
    public WaitCallback Action { get; set; }
    public object State { get; set; }

    public WorkItem(WaitCallback action, object state)
    {
        this.Action = action;
        this.State = state;
    }
}

然后来创建一个对象作为线程池中的一个线程:

public class SimpleThreadPoolThread
{
    private object locker = new object();
    private AutoResetEvent are = new AutoResetEvent(false);
    private WorkItem wi;
    private Thread t;
    private bool b = true;
    private bool isWorking;

    public bool IsWorking
    {
        get
        {
            lock (locker)
            {
                return isWorking;
            }
        }
    }
    public event Action<SimpleThreadPoolThread> WorkComplete;

    public SimpleThreadPoolThread()
    {
        lock (locker)
        {
            // 当前没有实际任务
            isWorking = false;
        }
        t = new Thread(Work) { IsBackground = true };
        t.Start();
    }

    public void SetWork(WorkItem wi)
    {
        this.wi = wi;
    }

    public void StartWork()
    {
        // 发出信号
        are.Set();
    }

    public void StopWork()
    {
        // 空任务
        wi = null;
        // 停止线程循环
        b = false;
        // 发出信号结束线程
        are.Set();
    }

    private void Work()
    {
        while (b)
        {
            // 没任务,等待信号
            are.WaitOne();
            if (wi != null)
            {
                lock (locker)
                {
                    // 开始
                    isWorking = true;
                }
                // 执行任务
                wi.Action(wi.State);
                lock (locker)
                {
                    // 结束
                    isWorking = false;
                }
                // 结束事件
                WorkComplete(this);
            }
        }
    }

代码的细节可以看注释,对这段代码的整体结构作一个说明:

1) 由于这个线程是被线程池中任务复用的,所以线程的任务处于循环中,除非线程池打算回收这个线程,否则不会退出循环结束任务

2) 使用自动信号量让线程没任务的时候等待,由线程池在外部设置任务后发出信号来执行实际的任务,执行完毕后继续等待

3) 线程公开一个完成的事件,线程池可以挂接处理方法,在任务完成后更新线程池状态

4) 线程池中的所有线程都是后台线程

下面再来实现线程池:

public class SimpleThreadPool : IDisposable
{
    private object locker = new object();
    private bool b = true;
    private int minThreads;
    private int maxThreads;
    private int currentActiveThreadCount;
    private List<SimpleThreadPoolThread> simpleThreadPoolThreadList = new List<SimpleThreadPoolThread>();
    private Queue<WorkItem> workItemQueue = new Queue<WorkItem>();

    public int CurrentActiveThreadCount
    {
        get
        {
            lock (locker)
            {
                return currentActiveThreadCount;
            }
        }

    }

    public int CurrentThreadCount
    {
        get
        {
            lock (locker)
            {
                return simpleThreadPoolThreadList.Count;
            }
        }
    }

    public int CurrentQueuedWorkCount
    {
        get
        {
            lock (locker)
            {
                return workItemQueue.Count;
            }
        }
    }

    public SimpleThreadPool()
    {
        minThreads = 4;
        maxThreads = 25;
        Init();
    }

    public SimpleThreadPool(int minThreads, int maxThreads)
    {
        if (minThreads > maxThreads)
            throw new ArgumentException("minThreads > maxThreads", "minThreads,maxThreads");
        this.minThreads = minThreads;
        this.maxThreads = maxThreads;
        Init();
    }

    public void QueueUserWorkItem(WorkItem wi)
    {
        lock (locker)
        {
            // 任务入列
            workItemQueue.Enqueue(wi);
        }
    }

    private void Init()
    {
        lock (locker)
        {
            // 一开始创建最小线程
            for (int i = 0; i < minThreads; i++)
            {
                CreateThread();
            }
            currentActiveThreadCount = 0;
        }
        new Thread(Work) { IsBackground = true }.Start();
    }

    private SimpleThreadPoolThread CreateThread()
    {
        SimpleThreadPoolThread t = new SimpleThreadPoolThread();
        // 挂接任务结束事件
        t.WorkComplete += new Action<SimpleThreadPoolThread>(t_WorkComplete);
        // 线程入列
        simpleThreadPoolThreadList.Add(t);
        return t;
    }

    private void Work()
    {
        // 线程池主循环
        while (b)
        {
            Thread.Sleep(100);
            lock (locker)
            {
                // 如果队列中有任务并且当前线程小于最大线程
                if (workItemQueue.Count > 0 && CurrentActiveThreadCount < maxThreads)
                {
                    WorkItem wi = workItemQueue.Dequeue();
                    // 寻找闲置线程
                    SimpleThreadPoolThread availableThread = simpleThreadPoolThreadList.FirstOrDefault(t => t.IsWorking == false);
                    // 无则创建
                    if (availableThread == null)
                        availableThread = CreateThread();
                    // 设置任务
                    availableThread.SetWork(wi);
                    // 开始任务
                    availableThread.StartWork();
                    // 增加个活动线程
                    currentActiveThreadCount++;
                }
            }
        }
    }

    private void t_WorkComplete(SimpleThreadPoolThread t)
    {
        lock (locker)
        {
            // 减少个活动线程
            currentActiveThreadCount--;
            // 如果当前线程数有所富裕并且比最小线程多
            if ((workItemQueue.Count + currentActiveThreadCount) < minThreads && CurrentThreadCount > minThreads)
            {
                // 停止已完成的线程
                t.StopWork();
                // 从线程池删除线程
                simpleThreadPoolThreadList.Remove(t);
            }
        }
    }

    public void Dispose()
    {
        // 所有线程停止
        foreach (var t in simpleThreadPoolThreadList)
        {
            t.StopWork();
        }
        // 线程池主循环停止
        b = false;
    }
}

线程池的结构如下:

1) 在构造方法中可以设置线程池最小和最大线程

2) 维护一个任务队列和一个线程池中线程的列表

3) 初始化线程池的时候就创建最小线程数量定义的线程

4) 线程池主循环每20毫秒就去处理一次,如果有任务并且线程池还可以处理任务的话,先是找闲置线程,找不到则创建一个

5) 通过设置任务委托以及发出信号量来开始任务

6) 线程池提供了三个属性来查看当前活动线程数,当前总线程数和当前队列中的任务数

7) 任务完成的回调事件中我们判断如果当前线程有富裕并且比最小线程多则回收线程

8) 线程池是IDispose对象,在Dispose()方法中停止所有线程后停止线程池主循环

写一段代码来测试线程池:

using (SimpleThreadPool t = new SimpleThreadPool(2, 4))
{
    Stopwatch sw2 = Stopwatch.StartNew();
    for (int i = 0; i < 10; i++)
    {
        t.QueueUserWorkItem(new WorkItem((index =>
        {
            Console.WriteLine(string.Format("#{0} : {1} / {2}", Thread.CurrentThread.ManagedThreadId, DateTime.Now.ToString("mm:ss"), index));
            Console.WriteLine(string.Format("CurrentActiveThread: {0} / CurrentThread: {1} / CurrentQueuedWork: {2}", t.CurrentActiveThreadCount, t.CurrentThreadCount, t.CurrentQueuedWorkCount));
            Thread.Sleep(1000);
        }), i));
    }
    while (t.CurrentQueuedWorkCount > 0 || t.CurrentActiveThreadCount > 0)
    {
        Thread.Sleep(10);
    }
    Console.WriteLine("All work completed");
    Console.WriteLine(string.Format("CurrentActiveThread: {0} / CurrentThread: {1} / CurrentQueuedWork: {2}", t.CurrentActiveThreadCount, t.CurrentThreadCount, t.CurrentQueuedWorkCount));
    Console.WriteLine(sw2.ElapsedMilliseconds);
} 

代码中我们向线程池推入10个任务,每个任务需要1秒执行,任务执行前输出当前任务的所属线程的Id,当前时间以及状态值。然后再输出线程池的几个状态属性。主线程循环等待所有任务完成后再次输出线程池状态属性以及所有任务完成耗费的时间:

image

我们可以看到:

1) 线程池中的线程总数从2到4到2

2) 线程池中活动的线程数从2到4到0

3) 线程池中排队的任务数从9到0

4) 所有线程完成一共使用了3秒时间

相比.NET内置的线程池,性能虽然有0.5秒的提高(可以见前文,.NET线程池在创建新的线程之前会等待0.5秒左右的时间),但其实一个好的线程池的实现需要考虑很多策略(什么时候去创建新线程,什么时候去回收老线程),.NET的ThreadPool在整体性能上做的很好,所以不建议随便去使用自定义的线程池。本例更只能作为实验和演示。

作者: lovecindywang
本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。
相关文章
|
2月前
|
Prometheus 监控 Cloud Native
JAVA线程池监控以及动态调整线程池
【10月更文挑战第22天】在 Java 中,线程池的监控和动态调整是非常重要的,它可以帮助我们更好地管理系统资源,提高应用的性能和稳定性。
217 64
|
21天前
|
NoSQL Redis
单线程传奇Redis,为何引入多线程?
Redis 4.0 引入多线程支持,主要用于后台对象删除、处理阻塞命令和网络 I/O 等操作,以提高并发性和性能。尽管如此,Redis 仍保留单线程执行模型处理客户端请求,确保高效性和简单性。多线程仅用于优化后台任务,如异步删除过期对象和分担读写操作,从而提升整体性能。
51 1
|
2月前
|
监控 安全 Java
在 Java 中使用线程池监控以及动态调整线程池时需要注意什么?
【10月更文挑战第22天】在进行线程池的监控和动态调整时,要综合考虑多方面的因素,谨慎操作,以确保线程池能够高效、稳定地运行,满足业务的需求。
123 38
|
19天前
|
Java API 调度
【JavaEE】——多线程(join阻塞,计算,引用,状态)
【JavaEE】——多线程,join,sleep引起的线程阻塞,多线程提升计算效率,如何获取线程的引用和状态
|
2月前
|
Java
.如何根据 CPU 核心数设计线程池线程数量
IO 密集型:核心数*2 计算密集型: 核心数+1 为什么加 1?即使当计算密集型的线程偶尔由于缺失故障或者其他原因而暂停时,这个额外的线程也能确保 CPU 的时钟周期不会被浪费。
83 4
|
3月前
|
存储 消息中间件 资源调度
C++ 多线程之初识多线程
这篇文章介绍了C++多线程的基本概念,包括进程和线程的定义、并发的实现方式,以及如何在C++中创建和管理线程,包括使用`std::thread`库、线程的join和detach方法,并通过示例代码展示了如何创建和使用多线程。
68 1
|
2月前
|
Java
线程池内部机制:线程的保活与回收策略
【10月更文挑战第24天】 线程池是现代并发编程中管理线程资源的一种高效机制。它不仅能够复用线程,减少创建和销毁线程的开销,还能有效控制并发线程的数量,提高系统资源的利用率。本文将深入探讨线程池中线程的保活和回收机制,帮助你更好地理解和使用线程池。
119 2
|
2月前
|
Prometheus 监控 Cloud Native
在 Java 中,如何使用线程池监控以及动态调整线程池?
【10月更文挑战第22天】线程池的监控和动态调整是一项重要的任务,需要我们结合具体的应用场景和需求,选择合适的方法和策略,以确保线程池始终处于最优状态,提高系统的性能和稳定性。
420 2
|
3月前
|
Java 开发者
在Java多线程编程中,创建线程的方法有两种:继承Thread类和实现Runnable接口
【10月更文挑战第20天】在Java多线程编程中,创建线程的方法有两种:继承Thread类和实现Runnable接口。本文揭示了这两种方式的微妙差异和潜在陷阱,帮助你更好地理解和选择适合项目需求的线程创建方式。
47 3
|
3月前
|
Java 开发者
在Java多线程编程中,选择合适的线程创建方法至关重要
【10月更文挑战第20天】在Java多线程编程中,选择合适的线程创建方法至关重要。本文通过案例分析,探讨了继承Thread类和实现Runnable接口两种方法的优缺点及适用场景,帮助开发者做出明智的选择。
29 2