自动化控制之线程池的使用

Posted 2020-09-29 天道酬勤，商道酬信。

说的线程池是一个类我总觉得这样叫有点不名副其实。其实就是一个类内部实现了FIFO队列，把临时数据放到这个队列里，“线程池类”内容按照入队的先后次序触发一个负责解析校验等的事件，并且把数据传递个这个事件。

昨天反编译看了公司项目首席架构师实现的线程池。非常之惊讶，在没有.net 4可用的年代。那思想和那技术是相当的可以啊。佩服。

这里说的线程池是一个类我总觉得这样叫有点不名副其实。其实就是一个类内部实现了FIFO队列，把临时数据放到这个队列里，“线程池类”按照入队的先后次序触发一个负责解析校验等的事件，并且把数据传递个这个事件。

好了，上代码：

    /// <summary>
    /// 自定义线程池类，不依赖.net Queue实现了先进先出处理队列里数据
    /// </summary>
    public class CoreThreadPool : IDisposable
    {
        /// <summary>
        /// 队列元素申明
        /// </summary>
        [StructLayout(LayoutKind.Sequential)]
        private class PoolData
        {
            /// <summary>
            /// 外部要求放入队列的数据
            /// </summary>
            public object Data;
            /// <summary>
            /// 需要执行的命令(Exit/Command(自定义))
            /// </summary>
            public CoreThreadPool.PoolCommand Command;
            public PoolData()
            {
                this.Command = CoreThreadPool.PoolCommand.Exit;
            }
            public PoolData(object data)
            {
                this.Data = data;
                this.Command = CoreThreadPool.PoolCommand.Command;
            }
            public PoolData(CoreThreadPool.PoolCommand cmd)
            {
                this.Command = cmd;
            }
        }
        protected enum PoolCommand
        {
            Command,
            Exit
        }
        protected SafeFileHandle complatePort;
        /// <summary>
        /// 线程池主线程
        /// </summary>
        protected Thread thread;
        protected volatile bool isOpened;
        [method: CompilerGenerated]
        [CompilerGenerated]
        public event Action<object> Exceute;
        [method: CompilerGenerated]
        [CompilerGenerated]
        public event Action<object> ExitExceute;
        /// <summary>
        /// 线程池是否正在运行
        /// </summary>
        public bool IsOpened
        {
            get
            {
                return this.isOpened;
            }
            set
            {
                this.isOpened = value;
            }
        }
        [DllImport("kernel32.dll", CharSet = CharSet.Auto, SetLastError = true)]
        private static extern SafeFileHandle CreateIoCompletionPort(IntPtr FileHandle, IntPtr ExistingCompletionPort, IntPtr CompletionKey, uint NumberOfConcurrentThreads);
        [DllImport("kernel32.dll", CharSet = CharSet.Auto, SetLastError = true)]
        private static extern bool GetQueuedCompletionStatus(SafeFileHandle CompletionPort, out uint lpNumberOfBytesTransferred, out IntPtr lpCompletionKey, out IntPtr lpOverlapped, uint dwMilliseconds);
        [DllImport("Kernel32", CharSet = CharSet.Auto)]
        private static extern bool PostQueuedCompletionStatus(SafeFileHandle CompletionPort, uint dwNumberOfBytesTransferred, IntPtr dwCompletionKey, IntPtr lpOverlapped);
        /// <summary>
        /// 启动线程池的主线程
        /// </summary>
        public void Start()
        {
            isOpened = true;
            if (thread != null)
            {
                throw new Exception("线程池已经是启动状态！");
            }
            complatePort = CreateIoCompletionPort(new IntPtr(-1), IntPtr.Zero, IntPtr.Zero, 0u);
            if (complatePort.IsInvalid)
            {
                throw new Exception(string.Format("创建IOCP出错！原因是:{0}", Marshal.GetLastWin32Error().ToString()));
            }
            thread = new Thread(new ParameterizedThreadStart(this.Run));
            thread.Start(complatePort);
        }
        /// <summary>
        /// 外部提交数据对象到队列
        /// </summary>
        /// <param name="data"></param>
        public void Post(object data)
        {
            PostData(new CoreThreadPool.PoolData(data));
        }
        /// <summary>
        /// 线程池主线程执行逻辑
        /// </summary>
        /// <param name="CompletionPortID"></param>
        private void Run(object CompletionPortID)
        {
            SafeFileHandle completionPort = (SafeFileHandle)CompletionPortID;            
            while (IsOpened)
            {
                uint num;
                IntPtr intPtr;
                IntPtr value;
                //从队列里取出最前面的对象
                CoreThreadPool.GetQueuedCompletionStatus(completionPort, out num, out intPtr, out value, 4294967295u);
                if (num > 0u)
                {
                    GCHandle gCHandle = GCHandle.FromIntPtr(value);
                    CoreThreadPool.PoolData poolData = (CoreThreadPool.PoolData)gCHandle.Target;
                    gCHandle.Free();
                    if (poolData.Command != CoreThreadPool.PoolCommand.Command)
                    {
                        IsOpened = false;
                        break;
                    }
                    RaiseExecute(poolData.Data);
                }
            }
            RaiseExitExecute("线程池已经停止。");
            isOpened = false;
            thread = null;
        }
        /// <summary>
        /// 触发Execute事件
        /// </summary>
        /// <param name="data"></param>
        private void RaiseExecute(object data)
        {
            Exceute?.Invoke(data);
        }
        /// <summary>
        /// 触发ExitExecute事件
        /// </summary>
        /// <param name="data"></param>
        private void RaiseExitExecute(object data)
        {
            ExitExceute?.Invoke(data);
        }
        /// <summary>
        /// 结束线程池主线程
        /// </summary>
        public void Stop()
        {
            PostData(new PoolData(PoolCommand.Exit));
            IsOpened = false;
        }
        /// <summary>
        /// 内部提交数据到线程池队列中
        /// </summary>
        /// <param name="data"></param>
        private void PostData(PoolData data)
        {
            if (complatePort.IsClosed)
            {
                return;
            }
            GCHandle value = GCHandle.Alloc(data);
            PostQueuedCompletionStatus(complatePort, (uint)IntPtr.Size, IntPtr.Zero, GCHandle.ToIntPtr(value));
        }
        public void Dispose()
        {
            if (this.thread != null && this.thread.ThreadState != System.Threading.ThreadState.Stopped)
            {
                this.Stop();
            }
        }
    }

　　代码亮点：队列是使用操作系统的，使用windows api实现的。牛吧。

由于现在项目已经依赖.net 4了。于是进行了模仿，经过一番测试，发现差不多，不过还是觉得在多线程环境下使用 ConcurrentQueue会更好些呢。

 /// <summary>
    /// 自定义线程池类，使用ConcurrentQueue实现了先进先出处理队列里数据
    /// </summary>
    public class CoolThreadPool<T>
    {
        protected ConcurrentQueue<T> queue = new ConcurrentQueue<T>();
        protected Thread thread;
        private volatile bool isOpened;
        public bool IsOpened
        {
            get
            {
                return isOpened;
            }
        }

        public event Action<T> Exceute;
        public event Action StopedExceute;
        /// <summary>
        /// 启动线程池的主线程
        /// </summary>
        public void Start()
        {
            if (thread != null)
            {
                throw new Exception("线程池已经是启动状态！");
            }
            thread = new Thread(Run);
            thread.Start();
            isOpened = thread != null;
        }

        /// <summary>
        /// 线程池主线程执行逻辑
        /// </summary>
        private void Run()
        {
            while (isOpened)
            {
                T temp;
                queue.TryDequeue(out temp);
                if (temp != null)
                {
                    RaiseExecute(temp);
                }
                else break;
            }
            isOpened = false;
            thread = null;
            RaiseStopedExceute();
        }

        /// <summary>
        /// 触发Execute事件
        /// </summary>
        /// <param name="data"></param>
        private void RaiseExecute(T data)
        {
            Exceute?.Invoke(data);
        }

        /// <summary>
        /// 触发停止Execute事件
        /// </summary>
        /// <param name="data"></param>
        private void RaiseStopedExceute()
        {
            StopedExceute?.Invoke();
        }

        /// <summary>
        /// 结束线程池主线程
        /// </summary>
        public void Stop()
        {
            PostData(default(T));
            isOpened = false;
        }

        /// <summary>
        /// 外部提交数据对象到队列
        /// </summary>
        /// <param name="data"></param>
        public void Post(T data)
        {
            PostData(data);
        }

        /// <summary>
        /// 内部提交数据到线程池队列中
        /// </summary>
        /// <param name="data"></param>
        private void PostData(T data)
        {
            queue.Enqueue(data);
        }

        public void Dispose()
        {
            if (this.thread != null && this.thread.ThreadState != System.Threading.ThreadState.Stopped)
            {
                this.Stop();
            }
        }
    }

测试代码：

string[] temp = new string[]
            {
                "This is 1.",
                "This is 2.",
                "This is 3.",
                "This is 4.",
                "This is 5.",
                "This is 6.",
                "This is 7.",
                "This is 8.",
                "This is 9.",
                "This is 10.",
                "This is 11.",
                "This is 12."
            };
            pool.Exceute += Pool_Exceute1;
            pool.Start();
            foreach (string data in temp)
            {
                pool.Post(data);
            }