MES-ETL/ConsoleApp2/SimulationService/SimulationInputService.cs

using ConsoleApp2.Const;
using ConsoleApp2.Helpers;
using ConsoleApp2.HostedServices;
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Text.RegularExpressions;

namespace ConsoleApp2.SimulationService
{

    public class SimulationInputService : IInputService
    {
        private readonly ILogger _logger;
        private readonly IOptions<DataInputOptions> _dataInputOptions;

        private readonly ProcessContext _context;

        public SimulationInputService(ILogger<InputService> logger,
            IOptions<DataInputOptions> dataInputOptions,
            ProcessContext context)
        {
            _logger = logger;
            _dataInputOptions = dataInputOptions;
            _context = context;
        }
        public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, ProcessContext context, CancellationToken cancellationToken)
        {
            var inputDir = _dataInputOptions.Value.InputDir;
            _logger.LogInformation("***** simulation input service start, working dir: {InputDir}, thread id: {ThreadId} *****", inputDir, Environment.CurrentManagedThreadId);
            var files = Directory.GetFiles(inputDir);
            if (files.Length == 0)
            {
                _logger.LogInformation("No source files found in {InputDir}", inputDir);
                return;
            }
            foreach (var tableName in tasksOptions.TableInfoConfig.Keys)
            {
                _logger.LogInformation("Working table: {tableName}", tableName);

                var dataCount = tasksOptions.TableInfoConfig[tableName].SimulaRowCount;//当前表要生成的总数据量
                var companyTotallCount = 1000;//当前表每个公司生成的总数据量
                var tempRecords = new List<DataRecord>();
                var sk = DataHelper.shareKeys.First();
                var companyID = DataHelper.companyIds.First();

                var shareKeyInterval = 20000;//每个sharekey的数据量 
                var getShareKeyTimes = 0;//sharekey生成的次数,每生成一次，改变sharekey的值
                var getCompanyIDTimes = 0;//公司生成的次数,每生成一次，改变companyID的值
                var shareKeyIntervalCount = 0;

                var source = _dataInputOptions.Value.CreateSource?.Invoke(tableName);
                var testRecord = await source.GetTestRecord();
                for (long i = 1; i <= dataCount; i++)
                {
                    shareKeyIntervalCount++;
                    if (shareKeyIntervalCount > shareKeyInterval)
                    {
                        sk = DataHelper.GetShareKey(getShareKeyTimes);
                        getShareKeyTimes++;
                        shareKeyIntervalCount = 0;
                    }
                    var fields = new string[testRecord.Fields.Length];
                    Array.Copy(testRecord.Fields, fields, testRecord.Fields.Length);
                    var record = new DataRecord(fields, testRecord.TableName, testRecord.Headers, companyID);
                    //更新record的ID、OrderNo,ShardKey值
                    if (record.Headers.Contains("ID"))
                    {
                        var index = Array.IndexOf(record.Headers, "ID");
                        if (index > -1)
                        {
                            record.Fields[index] = i.ToString();
                        }
                    }
                    if (record.TableName == "order_box_block" && record.Headers.Contains("BoxID"))
                    {
                        var index = Array.IndexOf(record.Headers, "BoxID");
                        if (index > -1)
                        {
                            record.Fields[index] = i.ToString();
                        }
                    }
                    if ((record.TableName == "order_block_plan_item" || record.TableName == "order_package_item") && record.Headers.Contains("ItemID"))
                    {
                        var index = Array.IndexOf(record.Headers, "ItemID");
                        if (index > -1)
                        {
                            record.Fields[index] = i.ToString();
                        }
                    }
                    if (record.TableName == "order" && record.Headers.Contains("OrderNo"))
                    {
                        var index = Array.IndexOf(record.Headers, "OrderNo");
                        if (index > -1)
                        {
                            record.Fields[index] = i.ToString();
                        }
                    }
                    if (record.Headers.Contains("ShardKey"))
                    {
                        var index = Array.IndexOf(record.Headers, "ShardKey");
                        if (index > -1)
                        {
                            record.Fields[index] = sk.ToString();
                        }
                    }
                    tempRecords.Add(record);
                    if (tempRecords.Count >= companyTotallCount || i >= dataCount - 1)
                    {
                        foreach (var rc in tempRecords)
                        {
                            _context.AddInput();
                            if(_context.InputCount== 2000000)
                            {
                                var a = 1;
                            }
                            producerQueue.Enqueue(rc);
                            if (cancellationToken.IsCancellationRequested)
                                return;
                        }
                        tempRecords.Clear();
                        companyID = DataHelper.GetCompanyId(getCompanyIDTimes);
                        getCompanyIDTimes++;
                    }
                }
                _logger.LogInformation("table:'{tableName}' simulation input completed", tableName);
                //}
                //_logger.LogInformation("File '{File}' input completed", Path.GetFileName(sqlPath));
            }

            context.CompleteInput();
            _logger.LogInformation("***** Csv input service completed *****");
        }
    }
}