BeiJinSettleAccount/code/src/Modules/SettleAccount/src/SettleAccount.Application/Entities/BQ/VmiAsyncBalanceService.cs

using System;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Diagnostics;
using System.Linq;
using System.Linq.Dynamic.Core;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using EFCore.BulkExtensions;
using LinqToDB;
using LinqToDB.EntityFrameworkCore;
using Magicodes.ExporterAndImporter.Core.Extension;
using Microsoft.AspNetCore.Mvc;
using Microsoft.AspNetCore.SignalR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Omu.ValueInjecter;
using SettleAccount.Job.SignalR;
using SqlSugar;
using Volo.Abp;
using Volo.Abp.Application.Services;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Validation;
using Win.Sfs.SettleAccount.Entities.BQ.Vmi;
using Win.Sfs.Shared.RepositoryBase;

namespace Win.Sfs.SettleAccount.Entities.BQ
{
    /// <summary>
    /// 异步更新库存
    /// </summary>
    [Route("[controller]/[action]")]
    public class VmiAsyncBalanceService : Controller, IApplicationService, IJobService, ITransientDependency
    {
        private readonly IServiceProvider _serviceProvider;
        private readonly ILogger<VmiAsyncBalanceService> _logger;
        private static readonly SemaphoreSlim semaphoreSlim = new SemaphoreSlim(1, 1);

        public VmiAsyncBalanceService(IServiceProvider serviceProvider, ILogger<VmiAsyncBalanceService> logger)
        {
            this._serviceProvider = serviceProvider;
            this._logger = logger;
            LinqToDBForEFTools.Initialize();
        }

        /// <summary>
        /// 异步更新库存，手动测试
        /// </summary>
        /// <returns></returns>
        [HttpPost]
        public async Task<double> Test()
        {
            var now = DateTime.Now;
            await Invoke(_serviceProvider).ConfigureAwait(false);
            return (DateTime.Now - now).TotalMinutes;
        }

        [NonAction]
        [DisableValidation]
        public async Task Invoke(IServiceProvider serviceProvider)
        {
            await semaphoreSlim.WaitAsync().ConfigureAwait(false);
            try
            {
                await InvokeInternal(serviceProvider).ConfigureAwait(false);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        }

        private async Task InvokeInternal(IServiceProvider serviceProvider)
        {
            var batchSize = 10000;
            var fetchSize = 0;
            var connectionString = serviceProvider.GetRequiredService<IConfiguration>().GetConnectionString("SettleAccountService");
            Dictionary<string, VmiBalance> balanceCachedList = null;
            for (var i = 0; i < 1000; i++)
            {
                //1:新建 2：更新
                var balanceList = new List<Tuple<VmiBalance, int>>();
                var vmiReplenishedList = new List<VmiReplenished>();
                var sw = new Stopwatch();
                sw.Start();
                using var connection = new Microsoft.Data.SqlClient.SqlConnection(connectionString);
                connection.Open();
                using var transaction = connection.BeginTransaction();
                try
                {
                    var command = connection.CreateCommand();
                    command.Transaction = transaction;
                    var options = new DbContextOptionsBuilder<SettleAccountDbContext>().UseSqlServer(connection).Options;
                    using var context = new SettleAccountDbContext(options);
                    context.Database.UseTransaction(transaction);
                    var vmiMessageRepo = context.Set<VmiMessage>();
                    var vmiLogRepo = context.Set<VmiLog>();
                    var vmiBalanceRepo = context.Set<VmiBalance>();
                    //读取可消费消息列表
                    var messages = vmiMessageRepo.AsNoTracking().Where(o => !o.isConsumed).OrderBy(o => o.Number).Take(batchSize).ToList();
                    //没有可消费消息则返回
                    if (!messages.Any())
                    {
                        transaction.Commit();
                        break;
                    }
                    //缓存余额表
                    var sw2 = new Stopwatch();
                    sw2.Start();
                    Func<VmiBalance, string> keySelector = o => $"{o.DeliverBillType}.{o.CodeType}.{o.RealPartCode}.{o.VinCode}.{o.ErpToLoc}.{o.OrderNum}.{o.factory}.{o.Configcode}";
                    //if (balanceCachedList == null)
                    //{
                        balanceCachedList = vmiBalanceRepo.AsNoTracking().ToDictionary(keySelector);
                    //}
                    sw2.Stop();
                    this._logger.LogInformation($"缓存结束:{balanceCachedList.Count}条,耗时 {sw2.ElapsedMilliseconds / 1000} 秒,{sw2.ElapsedMilliseconds} 豪秒");
                    //设置数量为实际返回数量
                    fetchSize = messages.Count;
                    //反序列化获取库存事务
                    var list = messages.Select(o =>
                    {
                        var log = JsonSerializer.Deserialize<VmiLog>(o.Message);
                        var jsonElement = JsonSerializer.Deserialize<JsonElement>(o.Message);
                        var property = jsonElement.GetProperty("Id");
                        Console.WriteLine(jsonElement);
                        var id = Guid.Parse(property.GetString());
                        log.SetId(id);
                        return new LogToBalance { VmiMessage = o, Log = log, Table = $"Set_VmiLog_{log.ChangedTime.Year}_{(log.ChangedTime.Month - 1) / 3 + 1}" };
                    });
                    //获取分表名称
                    var tables = list.Select(o => o.Table).Distinct().ToList();
                    //创建分表
                    foreach (var table in tables)
                    {
                        command.CommandText = $"select OBJECT_ID('{table}', 'U')";
                        var result = command.ExecuteScalar().ToString();
                        if (result == string.Empty)
                        {
                            command.CommandText = $"select * into {table} from Set_VmiLog where 1=0;";
                            command.ExecuteNonQuery();
                            command.CommandText = $"alter table {table} add constraint PK_{table} primary key (Id);";
                            command.ExecuteNonQuery();
                        }
                    }
                    foreach (var item in list)
                    {
                        var message = item.VmiMessage;
                        message.isConsumed = true;
                        var log = item.Log;
                        //本地查找
                        var balance = balanceList.Select(o => o.Item1).FirstOrDefault(
                            o =>
                            //o.DeliverBillType == log.DeliverBillType &&
                            //o.CodeType == log.CodeType &&
                            o.RealPartCode == log.RealPartCode &&
                            o.VinCode == log.VinCode &&
                            o.ErpToLoc == log.ErpToLoc
                            //&&
                            //o.OrderNum == log.OrderNum &&
                            //o.factory == log.factory &&
                            //o.Configcode == log.Configcode
                            );
                        //数据库查找
                        if (balance == null)
                        {
                            balance = balanceCachedList.GetValueOrDefault($"{log.DeliverBillType}.{log.CodeType}.{log.RealPartCode}.{log.VinCode}.{log.ErpToLoc}.{log.OrderNum}.{log.factory}.{log.Configcode}");
                            //if (balance == null)
                            //{
                            //    balance = vmiBalanceRepo.AsNoTracking().FirstOrDefault(
                            //        // balanceCachedList.Find(
                            //        o => o.DeliverBillType == log.DeliverBillType &&
                            //        o.CodeType == log.CodeType &&
                            //        o.RealPartCode == log.RealPartCode &&
                            //        o.VinCode == log.VinCode &&
                            //        o.ErpToLoc == log.ErpToLoc &&
                            //        o.OrderNum == log.OrderNum &&
                            //        o.factory == log.factory &&
                            //        o.Configcode == log.Configcode);
                            //}
                        }
                        if (balance == null)
                        {//不存在库存记录
                         //新建库存记录
                            balance = new VmiBalance();
                            balanceList.Add(new Tuple<VmiBalance, int>(balance, 1));
                            if (log.LogType == VmiLogType.Type300)
                            {//反结算入库，重建库存
                                var logHistory = vmiLogRepo.AsNoTracking().FirstOrDefault(
                                    o => o.LogType == VmiLogType.Type100 &&
                                    o.DeliverBillType == log.DeliverBillType &&
                                    o.CodeType == log.CodeType &&
                                    o.RealPartCode == log.RealPartCode &&
                                    o.VinCode == log.VinCode &&
                                    o.ErpToLoc == log.ErpToLoc &&
                                    o.OrderNum == log.OrderNum &&
                                    o.factory == log.factory &&
                                    o.Configcode == log.Configcode);
                                if (logHistory != null)
                                {
                                    balance.InjectFrom(logHistory);
                                }
                                else
                                {
                                    balance.InjectFrom(log);
                                }
                            }
                            else
                            {
                                balance.InjectFrom(log);
                            }
                            balance.Qty = log.ChangedQty;
                        }
                        else
                        {//存在库存记录
                            if (!balanceList.Any(o => o.Item1.Id == balance.Id))
                            {
                                balanceList.Add(new Tuple<VmiBalance, int>(balance, 2));
                            }
                            var logType = log.LogType;
                            var currentQty = balance.Qty;// + log.ty
                            if (logType == VmiLogType.Type100)
                            {
                                //发运入库，负库存字段需要更新
                                if (balance.Qty < decimal.Zero)
                                {
                                    balance.InjectFrom(log);
                                }
                            }
                            else if (logType == VmiLogType.Type300)
                            {
                                //反结入库，只更新库存
                            }
                            else if (logType == VmiLogType.Type500)
                            {
                                //调整入库，更新库存和其他字段
                                balance.InjectFrom(log);
                            }
                            if (logType == VmiLogType.Type100 && balance.Qty < decimal.Zero)
                            {
                                //添加负库存补货记录
                                var log2 = new VmiReplenished();
                                log2.InjectFrom(log);
                                vmiReplenishedList.Add(log2);
                            }
                            // 更新库存
                            balance.Qty = currentQty + log.ChangedQty;
                        }
                    }
                    //更新事务分表
                    foreach (var item in tables)
                    {
                        var logs = list.Where(o => o.Table == item).Select(o => o.Log).ToList();
                        await context.BulkCopyAsync(new LinqToDB.Data.BulkCopyOptions { TableName = item }, logs).ConfigureAwait(false);
                    }
                    //批量插入负库存补货记录
                    await context.BulkInsertAsync(vmiReplenishedList).ConfigureAwait(false);
                    //批量插入库存余额
                    var addList = balanceList.Where(o => o.Item2 == 1 && o.Item1.Qty != decimal.Zero).Select(o => o.Item1).ToList();
                    await context.BulkInsertAsync(addList).ConfigureAwait(false);
                    //批量更新库存余额
                    var editList = balanceList.Where(o => o.Item2 == 2 && o.Item1.Qty != decimal.Zero).Select(o => o.Item1).ToList();
                    await context.BulkUpdateAsync(editList).ConfigureAwait(false);
                    //批量删除库存余额
                    var removeList = balanceList.Where(o => o.Item2 == 2 && o.Item1.Qty == decimal.Zero).Select(o => o.Item1).ToList();
                    await context.BulkDeleteAsync(removeList).ConfigureAwait(false);
                    //批量删除消息
                    await context.BulkDeleteAsync(messages).ConfigureAwait(false);
                    transaction.Commit();
                    ////更新余额缓存
                    //var sw3 = new Stopwatch();
                    //sw3.Start();
                    //addList.ForEach(o => balanceCachedList.Add(keySelector(o), o));
                    //removeList.ForEach(o => balanceCachedList.Remove(keySelector(o)));
                    //sw3.Stop();
                    //this._logger.LogInformation($"更新缓存:{balanceList.Count}条,耗时 {sw3.ElapsedMilliseconds / 1000} 秒,{sw3.ElapsedMilliseconds} 豪秒");
                }
                catch (Exception ex)
                {
                    this._logger.LogError(ex.ToString());
                    throw new UserFriendlyException(ex.ToString(), "500");
                }
                finally
                {
                    sw.Stop();
                    this._logger.LogInformation($"处理{fetchSize}条,耗时 {sw.ElapsedMilliseconds / 1000 / 60}分钟,{sw.ElapsedMilliseconds / 1000}秒");
                }
                try
                {
                    using var scope = serviceProvider.CreateScope();
                    using var connection2 = new SqlConnection(connectionString);
                    connection2.Open();
                    var command = connection2.CreateCommand();
                    command.CommandText = "SELECT count(*) FROM Set_VmiMessage WITH(NOLOCK)";
                    var count = Convert.ToInt64(command.ExecuteScalar().ToString());
                    scope.ServiceProvider.GetService<IHubContext<PageHub>>().Clients.All.ServerToClient("VmiBalance", count.ToString(), "");
                }
                catch (Exception)
                {
                    throw;
                }
            }
        }
    }

    public class LogToBalance
    {
        public VmiMessage VmiMessage { get; set; }
        public VmiLog Log { get; set; }
        public string Table { get; set; }
    }
}