自已寫的線程池 - s0611163－IT工程師數位筆記本

文章出處

C#有ThreadPool和Task，為什么還要自己寫線程池？我以前也沒想過自己寫線程池，都是用ThreadPool或Task，前段時間寫爬蟲，我想控制10個線程爬網頁、10個線程下載網頁上的圖片，不然的話因為網頁很多，圖片相對較少，可能大部分線程都在爬網頁，少量線程在下載圖片，這樣下載圖片的速度慢了，所以我想到了自己寫線程池MyThreadPool，再配合ThreadPool使用，精確控制爬網頁和下載圖片的線程數。不過這個線程池寫的比較簡單，缺點是線程池會瞬間創建最大數量的工作線程。

線程池類代碼：

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace Common.Utils
{
    /// <summary>
    /// 線程池
    /// </summary>
    public static class MyThreadPool
    {
        /// <summary>
        /// 最大工作線程數
        /// </summary>
        private static int m_WorkerThreads = 10;
        /// <summary>
        /// 線程隊列
        /// </summary>
        private static ConcurrentQueue<MyThread> m_ThreadQueue = new ConcurrentQueue<MyThread>();
        /// <summary>
        /// 任務隊列
        /// </summary>
        private static ConcurrentQueue<Tuple<MyAction, object>> m_Action = new ConcurrentQueue<Tuple<MyAction, object>>();

        /// <summary>
        /// 創建并啟動線程
        /// </summary>
        /// <param name="action"></param>
        /// <param name="obj"></param>
        public static void Start(MyAction action, object obj = null)
        {
            m_Action.Enqueue(new Tuple<MyAction, object>(action, obj));

            MyThread thread;
            if (m_ThreadQueue.Count < m_WorkerThreads)
            {
                thread = new MyThread();
                m_ThreadQueue.Enqueue(thread);
                thread.isWorker = true; //設置為工作線程
                thread.thread = new Thread(new ThreadStart(() =>
                {
                    Tuple<MyAction, object> tuple;
                    while (thread.isWorker) //如果是工作線程，則一直循環
                    {
                        if (m_Action.TryDequeue(out tuple)) //如果任務隊列中有任務，則取出任務執行
                        {
                            tuple.Item1(tuple.Item2); //執行任務
                        }
                        Thread.Sleep(1);
                    }
                }));
                thread.thread.IsBackground = true;
                thread.thread.Start();
            }
        }

        /// <summary>
        /// 設置最大工作線程數
        /// </summary>
        /// <param name="workerThreads">最大工作線程數</param>
        public static void SetMaxThreads(int workerThreads)
        {
            m_WorkerThreads = workerThreads;
            MyThread thread = null;
            while (m_ThreadQueue.Count > m_WorkerThreads)
            {
                m_ThreadQueue.TryDequeue(out thread);
                thread.isWorker = false;
                Thread.Sleep(1);
            }
        }

        /// <summary>
        /// 獲取最大工作線程數
        /// </summary>
        public static int GetMaxThreads()
        {
            return m_WorkerThreads;
        }

        /// <summary>
        /// 獲取當前工作線程數
        /// </summary>
        public static int GetWorkerThreads()
        {
            return m_ThreadQueue.Count;
        }
    }

    /// <summary>
    /// 任務
    /// </summary>
    /// <param name="obj">任務參數</param>
    public delegate void MyAction(object obj);

    /// <summary>
    /// 線程
    /// </summary>
    public class MyThread
    {
        /// <summary>
        /// 線程
        /// </summary>
        public Thread thread { get; set; }
        /// <summary>
        /// 是否工作線程
        /// </summary>
        public bool isWorker { get; set; }
    }
}

View Code

測試代碼：

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;
using Common.Utils;
using System.Collections.Concurrent;

namespace test
{
    public partial class Form1 : Form
    {
        private static int errorCount = 0;
        private static int errorCount2 = 0;
        private static object _lock = new object();
        private static int dataCount = 30;
        /// <summary>
        /// 任務執行總次數
        /// </summary>
        private static int runCount = 0;

        public Form1()
        {
            InitializeComponent();
        }

        private void Form1_Load(object sender, EventArgs e)
        {
            ThreadPool.SetMaxThreads(20, 20);
            StringBuilder sb = new StringBuilder();
            for (int k = 1; k <= dataCount; k++)
            {
                sb.AppendFormat("{0},", k.ToString("00"));
            }
            lblMsg2.Text = sb.ToString();

            timer1.Tick += new EventHandler((obj, ea) =>
            {
                int maxThreads = MyThreadPool.GetMaxThreads();
                int workerThreads = MyThreadPool.GetWorkerThreads();
                lblMsg.Text = string.Format("{0}/{1}(工作線程數/最大工作線程數)", workerThreads, maxThreads);
                //ThreadPool.GetMaxThreads(out workerThreads, out maxThreads);
                //lblMsg.Text = string.Format("{0}/{1}(最大輔助線程數/最大I/O線程數)", workerThreads, maxThreads);

                if (workerThreads > maxThreads)
                {
                    errorCount++;
                    lblError.Text = "錯誤(工作線程數超過最大工作線程數)次數：" + errorCount.ToString() + "，錯誤(線程內代碼執行結果錯誤)次數：" + errorCount2.ToString();
                }
                if (lblMsg2.Text != sb.ToString())
                {
                    errorCount2++;
                    lblError.Text = "錯誤(工作線程數超過最大工作線程數)次數：" + errorCount.ToString() + "，錯誤(線程內代碼執行結果錯誤)次數：" + errorCount2.ToString();
                    LogUtil.LogError(lblMsg2.Text);
                }
                lblRunCount.Text = "任務執行總次數：" + runCount.ToString();
            });
            timer1.Interval = 100;
            timer1.Start();
            lblError.Text = "暫沒有錯誤";
        }

        //開始
        private void button1_Click(object sender, EventArgs e)
        {
            MyThreadPool.SetMaxThreads(20);
            ThreadPool.SetMaxThreads(20, 20);
            button1.Enabled = false;
            Thread thread = new Thread(new ThreadStart(() =>
            {
                while (true)
                {
                    List<int> list = new List<int>();
                    for (int i = 1; i <= dataCount; i++)
                    {
                        // ThreadPool.QueueUserWorkItem((obj) =>
                        // Task.Factory.StartNew((obj) =>
                        MyThreadPool.Start((obj) =>
                        {
                            int n = (int)obj;
                            lock (_lock)
                            {
                                list.Add(n);
                            }
                            if (list.Count == dataCount)
                            {
                                list.Sort();
                                StringBuilder sb = new StringBuilder();
                                for (int k = 0; k < list.Count; k++)
                                {
                                    sb.AppendFormat("{0},", list[k].ToString("00"));
                                }
                                this.Invoke(new MyInvoke(() => { lblMsg2.Text = sb.ToString(); }));
                                runCount++;
                            }
                        }, i);
                    }
                    Thread.Sleep(1);
                }
            }));
            thread.IsBackground = true;
            thread.Start();
        }

        //設置最大工作線程數
        private void button2_Click(object sender, EventArgs e)
        {
            int d;
            if (int.TryParse(txtMaxThreads.Text, out d))
            {
                if (d > 0)
                {
                    MyThreadPool.SetMaxThreads(d);
                    ThreadPool.SetMaxThreads(d, d);
                }
                else
                {
                    txtMaxThreads.Text = "1";
                }
            }
            else
            {
                txtMaxThreads.Text = "20";
            }
        }

    }

    public delegate void MyInvoke();
}