生产消费模式：多线程读写队列ConcurrentQueue

Posted 2020-10-17 二奎

需求：现需要将多个数据源的数据导入到目标数据库，这是一个经典的生产消费应用的例子。

直接上代码，看下实现：

            // 初始化列队缓冲区 队列大小为100
            IDataCollection<List<T>> queue = new QueueCollection<List<T>>(100);

            //开启X个后台任务，读取RabbitMQ队列信息, 把列队信息插入缓冲区队列
            var count = 1;
            for (int i = 0; i < count; i++)
            {
                Task.Factory.StartNew(() => new Producer<List<T>>(queue).Start(new RabbitSource<List<T>>().Get));
            }

            //开启X个后台任务，主动获取数据库数据，作为数据生产者，插入到缓冲区队列，
            for (int i = 0; i < count; i++)
            {
                Task.Factory.StartNew(() => new Producer<List<T>>(queue).Start(new DatabaseSource<List<T>>().Get));
            }

            //开启X个后台任务，主动获取读取缓冲区列队，作为数据消息者，把数据插入到ES库，
            for (int i = 0; i < count; i++)
            {
                Task.Factory.StartNew(() => new Customer<List<T>>(queue).Start(new Elastic().Insert));
            }

队列我们采用线程安全的ConcurrentQueue队列：

/// <summary>
    /// 缓冲区队列
    /// ConcurrentQueue线程安全，不用考虑锁的问题
    /// </summary>
    public class QueueCollection<T> :IDataCollection<T>
    {
        //队列最大值
        private int _maxSize;

        /// <summary>
        /// 线程安全的队列
        /// </summary>
        private ConcurrentQueue<T> _queue;

        public QueueCollection(int maxSize)
        {
            this._maxSize = maxSize;
            _queue = new ConcurrentQueue<T>();
        }
        
        public bool isPopWaiting()
        {
            return !_queue.Any();
        }

        public bool isPushWaiting()
        {
            return this._maxSize == _queue.Count;
        }
        
        public T Pop()
        {
            T _obj = default(T);
            if (!_queue.IsEmpty)
                _queue.TryDequeue(out _obj);

            return _obj;
        }

        public void Push(T t)
        {
            if (this._maxSize > _queue.Count)
            {
                _queue.Enqueue(t);
            }
        }
    }

如果我们不使用这个队列，只要满足IDataCollection接口，也可以进行替换：

public interface IDataCollection<T>
    {
        /// <summary>
        /// 插入数据 
        /// </summary>
        /// <param name="t"></param>
        void Push(T t);

        /// <summary>
        /// 取出数据
        /// </summary>
        /// <returns></returns>
        T Pop();

        /// <summary>
        /// 是否插入数据等待
        /// </summary>
        /// <returns></returns>
        bool isPushWaiting();

        /// <summary>
        /// 是否取出数据等待
        /// </summary>
        /// <returns></returns>
        bool isPopWaiting();
        
    }

生产者：

 public class Producer<T> : ITransientDependency
    {
        private int sleep;

        private IDataCollection<T> bufferQueue;

        public Producer(IDataCollection<T> queue)
        {
            sleep = 3000;
            bufferQueue = queue;
        }

        public void Start(Action<Action<T>> methodCall)
        {
            //入队
            methodCall((bills) => 
            {
                this.Enqueue(bills);
            });
        }

        private void Enqueue(T t)
        {
            var isWaiting = true;

            while (isWaiting)
            {
                if (!bufferQueue.isPushWaiting())
                {
                    this.bufferQueue.Push(t);
                    isWaiting = false;
                }
                else
                {
                    //生产者等待时间
                    Thread.Sleep(sleep);
                }
            }
        }
    }

消费者：

/// <summary>
    /// 消费者
    /// </summary>
    public class Customer<T>
    {
        //产品缓存队列
        private IDataCollection<T> _queue;
        
        //消费者等待时间
        private int Spead = 5000;//消费者等待时间

        public Customer(IDataCollection<T> queue)
        {
            this._queue = queue;
        }

        public void Start(Action<T> method)
        {
            while (true)
            {
                if (!_queue.isPopWaiting())
                {
                    T box = this._queue.Pop();

                    method(box);
                }
                else
                {
                    Thread.Sleep(Spead);
                }
            }
        }
    }

方法委托，也写了个基类，其实意义并不大，只是为了规范, 防止方法命名随意起。

    public interface IDataSource<T>
    {
        void Get(Action<T> func);
    }

最后，在DataSource的get方法中，调用 func即可。