多線程數據寫入隊列，異步線程進行批量處理

設計目的：

在多線程環境中，多線程處理數據時，如果每線程都單獨寫數據庫，性能低下。因此，為提高性能，數據需批量寫到數據庫中。出於此目的，進行了數據隊列的設計：

 

實現代碼：

 

/// <summary>
    /// 多線程異步推送數據，一線程異步批量處理數據
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public abstract class SingleThreadQueue<T>
    {
        protected List<T> queue;
        private Action _beforAction;//在批量執行前執行的方法
        private Action _afterAction;//在批量執行完成后執行的方法  
        private bool _runing = false;
        private int _maxCount, _maxSize;

        /// <summary>
        /// 
        /// </summary> 
        /// <param name="maxCount">每次最大處理數</param>
        /// <param name="maxSize">緩沖最大數</param>
        /// <param name="beforAction">開始執行處理前的動作</param>
        /// <param name="afterAction">所有數據處理完成后的動作</param>
        public SingleThreadQueue(int maxCount, int maxSize, Action beforAction = null, Action afterAction = null)
        {
            this.queue = new List<T>();
            this._beforAction = beforAction;
            this._afterAction = afterAction;
            this._maxCount = maxCount;
            this._maxSize = maxSize;
        }

        /// <summary>
        /// 異步推送數據
        /// </summary>
        /// <param name="item"></param>
        /// <returns>數據成功加入待處理隊列，返回true,數據達到隊列的最大緩沖，數據不會進入隊列，返回false</returns>
        public bool AsyncEnqueue(T item)
        {
            lock (this)
            {
                if (this.queue.Count >= this._maxSize)
                    return false;

                this.queue.Add(item);
                this.Activate();
                return true;
            }
        }

        protected T[] DoDequeue(List<T> queue)
        {
            if (queue.Count > this._maxCount)
            {
                var ds = queue.Take(this._maxCount).ToArray();
                queue.RemoveRange(0, this._maxCount);
                return ds;
            }
            else
            {
                var ds = queue.ToArray();
                queue.Clear();
                return ds;
            }
        }

        /// <summary>
        /// 實現此方法，批量執行處理時的方法
        /// </summary>
        /// <param name="items"></param>
        protected abstract void OnExecute(T[] items);

        private void Activate()
        {
            if (this._runing)
            {
                return;
            }

            this._runing = true;
            ThreadPool.QueueUserWorkItem((obj) =>
            {
                try
                {
                    this._beforAction?.Invoke();

                    T[] items;
                    //管理線程
                    while (true)
                    {
                        lock (this)
                        {
                            if (queue.Count < 1)
                            {
                                this._afterAction?.Invoke();
                                this._runing = false;
                                break;
                            }

                            items = this.DoDequeue(queue);
                        }

                        try
                        {
                            this.OnExecute(items);
                        }
                        catch (Exception ex)
                        {
                            //TODO:異常預警
                            // Loger.Exception(ex.Message, ex);
                        }
                    }
                }
                catch (Exception ex)
                {
                    //TODO:異常預警
                    // Loger.Exception(ex.Message, ex);
                }
            });
        }

    }

 經測試，多線程單獨寫入DB，在開發環境，100/秒，但批量處理可達5000/秒。