.NET Core TPL 数据流 BatchBlock管道封装

场景：例如需要处理数据库大量的数据，先查询出来，然后操作数据，然后1000条合成1批次，然后再插入到另外一张表。

作用：数据流批处理，多核操作增加效率，批处理改变单个插入为批量插入增加效率

定义数据Model 

    public class Employee
    {
        public int EmployeeID { get; set; }
        public string LastName { get; set; }
        public string FirstName { get; set; }

        // A random number generator that helps tp generate
        // Employee property values.
        static Random rand = new Random(42);

        // Possible random first names.
        static readonly string[] firstNames = { "Tom", "Mike", "Ruth", "Bob", "John" };
        // Possible random last names.
        static readonly string[] lastNames = { "Jones", "Smith", "Johnson", "Walker" };

        // Creates an Employee object that contains random
        // property values.
        public static Employee Random()
        {
            return new Employee
            {
                EmployeeID = -1,
                LastName = lastNames[rand.Next() % lastNames.Length],
                FirstName = firstNames[rand.Next() % firstNames.Length]
            };
        }
    }

定义数据库操作（这里简化）

    public class EmployeeRepository
    {
        /// <summary>
        /// 执行次数
        /// </summary>
        private static int Count;
        public static void InsertEmployees(Employee[] employees)
        {
            Console.WriteLine($"线程Id：{Thread.CurrentThread.ManagedThreadId} 第{Interlocked.Increment(ref Count)}次  总数：{ employees.Count()},数据：{JsonConvert.SerializeObject(employees.Select(p=>p.EmployeeID))}开始执行批量插入");
            //todo: db op
            return;
        }
    }

数据流批处理封装类

    public class BatchBlockPipeline<T>
    {
        /// <summary>
        /// 批处理块
        /// </summary>
        private BatchBlock<T> _batchBlock;
        /// <summary>
        /// 批处理执行块
        /// </summary>
        private ActionBlock<T[]> _actionBlock;
        /// <summary>
        /// 是否为定时触发
        /// </summary>
        private bool _timeTrigger;
        /// <summary>
        /// 定时触发时候用到的连接块
        /// </summary>
        private TransformBlock<T, T> _transformBlock;
        /// <summary>
        /// 定时触发器
        /// </summary>
        private readonly Timer _timer;

        /// <summary>
        /// 基本构造函数
        /// </summary>
        /// <param name="batchSize">每次处理的数据量</param>
        /// <param name="action">执行委托方法</param>
        /// <param name="boundedCapacity">最大处理的数据量 默认 int.MaxValue 2147483647</param>
        /// <param name="maxDegreeOfParallelism">最大并行量 默认1</param>
        /// <param name="timeTrigger">定时触发批处理 默认不处理， 设置大于0则处理，秒级别</param>
        public BatchBlockPipeline(int batchSize, Action<T[]> action, int boundedCapacity = int.MaxValue, int maxDegreeOfParallelism = 1, int timeTrigger = 0)
        {
            _batchBlock = new BatchBlock<T>(batchSize, new GroupingDataflowBlockOptions() { BoundedCapacity = boundedCapacity });
            _actionBlock = new ActionBlock<T[]>(data => action(data), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism });
            _batchBlock.LinkTo(_actionBlock, new DataflowLinkOptions() { PropagateCompletion = true });
            _batchBlock.Completion.ContinueWith(delegate { _actionBlock.Complete(); });
            if (timeTrigger > 0)
            {
                _timeTrigger = true;
                _transformBlock = new TransformBlock<T, T>(model =>
                {
                    _timer.Change(TimeSpan.FromSeconds(timeTrigger), Timeout.InfiniteTimeSpan);
                    return model;
                }, new ExecutionDataflowBlockOptions() { BoundedCapacity = boundedCapacity });
                _transformBlock.LinkTo(_batchBlock, new DataflowLinkOptions() { PropagateCompletion = true });
            }
        }

        /// <summary>
        /// post 数据
        /// </summary>
        /// <param name="model"></param>
        /// <returns></returns>
        public bool PostValue(T model)
        {
            if (!_timeTrigger)
            {
                return _batchBlock.Post(model);
            }
            return _transformBlock.Post(model);
        }

        /// <summary>
        /// 主动触发数据处理，例如：当数据剩余未达到batchsize 主动触发处理数据
        /// </summary>
        /// <param name="model"></param>
        /// <returns></returns>
        public void TriggerBatch()
        {
            _batchBlock.TriggerBatch();
        }

        /// <summary>
        /// 返回当前执行总数
        /// </summary>
        /// <returns></returns>
        public int GetBatchSum()
        {
            return _batchBlock.Receive().Count();
        }

        /// <summary>
        /// 主动关闭
        /// </summary>
        /// <returns></returns>
        public void Close()
        {
            if (!_timeTrigger)
            {
                _batchBlock.Complete();
            }
            _transformBlock.Complete();
        }
    }

测试方法

    class Program
    {
        static void Main(string[] args)
        {

            var batchDataPipeline = new BatchBlockPipeline<Employee>(10, EmployeeRepository.InsertEmployees);

            for (int i = 0; i < 100; i++)
            {
                batchDataPipeline.PostValue(Employee.Random(i));
            }

            Console.ReadKey();
        }
    }

1.测试：100条数据，10个一批次插入数据库，并行数量1

//var batchDataPipeline = new BatchBlockPipeline<Employee>(10, EmployeeRepository.InsertEmployees);

2.测试：100条数据，10个一批次插入数据库，并行数量1，最大限制处理20个

// var batchDataPipeline = new BatchBlockPipeline<Employee>(10, EmployeeRepository.InsertEmployees,20);

 3.测试 100条数据，10个一批次插入数据库，多核多线程并行数量10

//var batchDataPipeline = new BatchBlockPipeline<Employee>(10, EmployeeRepository.InsertEmployees, maxDegreeOfParallelism: 10);

 4.测试 100条数据，10个一批次插入数据库，定时触发3秒触发一次 （模拟不定期post数据，或者最后剩余的数据达不到batchsize的情况）

//var batchDataPipeline = new BatchBlockPipeline<Employee>(10, EmployeeRepository.InsertEmployees, timeTrigger: 3);

//for (int i = 0; i < 100; i++)
//{
// batchDataPipeline.PostValue(Employee.Random(i));
// if (i % 3 == 0|| i % 7 == 0 || i % 9 == 0)
// {
// Thread.Sleep(5000);
// }
//}

官方TPL数据流之BatchBlock例子链接     DemoGitHub地址