kxs-stat/AppStart/Helper/SplitTableData/AddRecordService.cs

using System;
using System.Collections.Generic;
using Library;
using LitJson;
using System.Linq;
using System.Data;
using System.Threading;
using MySystem.PxcModels;

namespace MySystem
{
    public class AddRecordService
    {
        public readonly static AddRecordService Instance = new AddRecordService();
        private AddRecordService()
        { }

        //分表字段集合
        private Dictionary<string, string> FieldDic = new Dictionary<string, string>()
        {
            {"TradeRecord", "Id,Sort,QueryCount,Status,Version,CreateDate,UpdateDate,CreateMan,UpdateMan,SeoTitle,SeoKeyword,SeoDescription,MerUserId,TopUserId,Remark,BrandId,DirectFlag,CapFlag,QrPayFlag,BankCardType,TradeAmount,TradeSerialNo,ClearDate,TradeDate,SnNo,MerBuddyType,HelpMonthCount,MerHelpFlag,MerNo,MerchantId,UserId,RecordNo,ParentNav,PayType,VipFlag,ActStatus"},
            {"UserAccountRecord", "Id,Sort,QueryCount,Status,Version,CreateDate,UpdateDate,CreateMan,UpdateMan,SeoTitle,SeoKeyword,SeoDescription,TransRecordNo,Remark,AfterBalanceAmount,BeforeBalanceAmount,AfterFreezeAmount,BeforeFreezeAmount,AfterTotalAmount,BeforeTotalAmount,ChangeAmount,ProductType,ChangeType,UserId,Kind"},
            {"SpBindRecord", "Id,Sort,QueryCount,Status,CreateDate,UpdateDate,SeoTitle,SeoKeyword,SeoDescription,ChannelSerial,UpdateTime,CreateTime,Remark,MerNewSnNo,ProductType,MerStatus,MerOperateType,MerOperateDate,MerSnNo,MerName,AgentName,MerNo,Field5,Field4,Field3,Field2,Field1"},
            {"SpMerchants", "Id,Sort,QueryCount,Status,CreateDate,UpdateDate,SeoTitle,SeoKeyword,SeoDescription,AgentNo,SnNo,UpdateTime,CreateTime,Remark,City,Province,ProductType,MerStatus,MerAuditDate,MerRegDate,MerIdcardNo,MerMobile,MerRealName,AgentName,MerName,MerNo,Field5,Field4,Field3,Field2,Field1"},
            {"SpActivateRecord", "Id,Sort,QueryCount,Status,CreateDate,UpdateDate,SeoTitle,SeoKeyword,SeoDescription,ProductType,AgentNo,Remark,BizEnterName,BizEnterNo,MerIdcardNo,MerRealName,ActivateDate,ActivateStatus,AssessMonthCount,AssessMonth,MerRegDate,SnNo,MerMobile,MerName,MerNo,ChannelSerial,Field5,Field4,Field3,Field2,Field1"},
            {"SpTradeRecord", "Id,Sort,QueryCount,Status,CreateDate,UpdateDate,SeoTitle,SeoKeyword,SeoDescription,ChannelSerial,AgentNo,ProductType,Remark,SettleMethod,SettleFee,BankCardNo,ReceiptType,IsStoreCashier,DigAmt,DigAmtFlag,MerMobile,DeviceType,TradeSnNo,BankCardType,SerEntryMode,TradeType,TradeTime,TradeDate,ErrorMsg,ErrorCode,TradeStatus,DiscountRateFlag,BankAuthCode,TradeReferNo,TradeAmount,MerName,MerNo,TradeSerialNo,Field5,Field4,Field3,Field2,Field1"},
        };

        //交易记录队列
        public void StartTradeRecord()
        {
            Thread th = new Thread(StartTradeRecordDo);
            th.IsBackground = true;
            th.Start();
        }
        public void StartTradeRecordDo()
        {
            while (true)
            {
                string content = RedisDbconn.Instance.RPop<string>("AddTradeRecordQueue");
                if(!string.IsNullOrEmpty(content))
                {
                    string[] datalist = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
                    AddData(datalist[0], datalist[1], "TradeRecord");
                }
                else
                {
                    Thread.Sleep(10000);
                }
            }
        }

        //收支明细队列
        public void StartUserAccountRecord()
        {
            Thread th = new Thread(StartUserAccountRecordDo);
            th.IsBackground = true;
            th.Start();
        }
        public void StartUserAccountRecordDo()
        {
            while (true)
            {
                string content = RedisDbconn.Instance.RPop<string>("AddUserAccountRecordQueue");
                if(!string.IsNullOrEmpty(content))
                {
                    string[] datalist = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
                    AddData(datalist[0], datalist[1], "UserAccountRecord");
                }
                else
                {
                    Thread.Sleep(10000);
                }
            }
        }

        //SP绑定记录队列
        public void StartSpBindRecord()
        {
            Thread th = new Thread(StartSpBindRecordDo);
            th.IsBackground = true;
            th.Start();
        }
        public void StartSpBindRecordDo()
        {
            while (true)
            {
                string content = RedisDbconn.Instance.RPop<string>("AddSpBindRecordQueue");
                if(!string.IsNullOrEmpty(content))
                {
                    string[] datalist = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
                    AddData(datalist[0], datalist[1], "SpBindRecord");
                }
                else
                {
                    Thread.Sleep(10000);
                }
            }
        }

        //SP商户信息队列
        public void StartSpMerchants()
        {
            Thread th = new Thread(StartSpMerchantsDo);
            th.IsBackground = true;
            th.Start();
        }
        public void StartSpMerchantsDo()
        {
            while (true)
            {
                string content = RedisDbconn.Instance.RPop<string>("AddSpMerchantsQueue");
                if(!string.IsNullOrEmpty(content))
                {
                    string[] datalist = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
                    AddData(datalist[0], datalist[1], "SpMerchants");
                }
                else
                {
                    Thread.Sleep(10000);
                }
            }
        }

        //SP激活押金队列
        public void StartSpActivateRecord()
        {
            Thread th = new Thread(StartSpActivateRecordDo);
            th.IsBackground = true;
            th.Start();
        }
        public void StartSpActivateRecordDo()
        {
            while (true)
            {
                string content = RedisDbconn.Instance.RPop<string>("AddSpActivateRecordQueue");
                if(!string.IsNullOrEmpty(content))
                {
                    string[] datalist = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
                    AddData(datalist[0], datalist[1], "SpActivateRecord");
                }
                else
                {
                    Thread.Sleep(10000);
                }
            }
        }

        //SP交易记录队列
        public void StartSpTradeRecord()
        {
            Thread th = new Thread(StartSpTradeRecordDo);
            th.IsBackground = true;
            th.Start();
        }
        public void StartSpTradeRecordDo()
        {
            while (true)
            {
                string content = RedisDbconn.Instance.RPop<string>("AddSpTradeRecordQueue");
                if(!string.IsNullOrEmpty(content))
                {
                    string[] datalist = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
                    AddData(datalist[0], datalist[1], "SpTradeRecord");
                }
                else
                {
                    Thread.Sleep(10000);
                }
            }
        }


        //添加表数据
        public void AddData(string content, string date, string table)
        {
            try
            {
                if(string.IsNullOrEmpty(RedisDbconn.Instance.Get<string>(table + date)))
                {
                    string createSql = "";
                    if(table == "TradeRecord")
                    {
                        createSql = AppConfig.Base.CreateTradeRecord;
                    }
                    if(table == "UserAccountRecord")
                    {
                        createSql = AppConfig.Base.CreateUserAccountRecord;
                    }
                    CustomerSqlConn.op(createSql.Replace("#DateTime#", date), AppConfig.Base.StatSqlConn);
                    RedisDbconn.Instance.Set(table + date, "1");
                    int sec = 3600 * 25;
                    if(date.Length == 6)
                    {
                        sec = 3600 * 24 * 35;
                    }
                    RedisDbconn.Instance.SetExpire(table + date, sec);
                    Thread.Sleep(2000);
                }

                JsonData selfDr = JsonMapper.ToObject(content);
                string fields = FieldDic[table];
                string val = "";
                foreach(string field in fields.Split(','))
                {
                    if(selfDr[field] == null)
                    {
                        val += "null,";
                    }
                    else
                    {
                        JsonType t = selfDr[field].GetJsonType();
                        if(t == JsonType.Int || t == JsonType.Double || t == JsonType.Long)
                        {
                            val += selfDr[field].ToString() + ",";
                        }
                        else if(t == JsonType.None)
                        {
                            val += "null,";
                        }
                        else
                        {
                            val += "'" + selfDr[field].ToString() + "',";
                        }
                    }
                }
                string sql = "INSERT INTO " + table + date + " (" + fields + ") VALUES (" + val.TrimEnd(',') + ")";
                CustomerSqlConn.op(sql, AppConfig.Base.StatSqlConn);
            }
            catch (Exception ex)
            {
                function.WriteLog(DateTime.Now.ToString() + "\n" + ex.ToString(), "添加记录RDS异常");
            }
            function.WriteLog(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + "\n\n", "添加记录RDS");
        }

    }
}