通过MQ队列执行SP原始数据

AppStart/Helper/SpDataHelper.cs

@@ -0,0 +1,102 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.Text;
+using System.Threading;
+using LitJson;
+using MySystem;
+using MySystem.SpModels;
+
+//营业额日汇总统计
+public class SpDataHelper
+{
+    public readonly static SpDataHelper Instance = new SpDataHelper();
+    private SpDataHelper()
+    { }
+
+    public void Start(string QueueName)
+    {
+        RabbitMQClient2.Instance.StartReceive(QueueName, content => DoWorks(content));
+    }
+
+    public void DoWorks(string content)
+    {
+        try
+        {
+            if(!string.IsNullOrEmpty(content))
+            {
+                string[] data = content.Split(new string[]{ "#cut#" }, StringSplitOptions.None);
+                string tableName = data[0];
+                string jsonString = data[1];
+                if(tableName == "TradeRecord") DoQueueTrade(jsonString);
+            }
+        }
+        catch (Exception ex)
+        {
+            Utils.WriteLog(DateTime.Now.ToString() + "\n" + ex, "SP原始数据入库异常");
+        }
+    }
+
+    public BlockingCollection<TradeRecord> list = new BlockingCollection<TradeRecord>();
+    public void DoQueueTrade(string jsonString)
+    { 
+        TradeRecord trade = Newtonsoft.Json.JsonConvert.DeserializeObject<TradeRecord>(jsonString);
+        list.Add(trade);
+    }
+
+
+
+    public void StartData()
+    {
+        Thread th = new Thread(StartDataDo);
+        th.IsBackground = true;
+        th.Start();
+    }
+
+    public void StartDataDo()
+    {
+        List<string> BrandIds = new List<string>();
+        BrandIds.Add("14");
+        BrandIds.Add("17");
+        BrandIds.Add("23");
+        BrandIds.Add("24");
+        BrandIds.Add("25");
+        BrandIds.Add("26");
+        BrandIds.Add("32");
+        MySystem.PxcModels.WebCMSEntities db = new MySystem.PxcModels.WebCMSEntities();
+        MySystem.JavaProductModels.WebCMSEntities pdb = new MySystem.JavaProductModels.WebCMSEntities();
+        bool conn = true;
+        while (true)
+        {
+            try
+            {
+                TradeRecord trade;
+                if(list.TryTake(out trade))
+                {
+                    if(!conn)
+                    {
+                        db = new MySystem.PxcModels.WebCMSEntities();
+                        pdb = new MySystem.JavaProductModels.WebCMSEntities();
+                        conn = true;
+                    }
+                    Utils.WriteLog("", "SP原始数据队列日志");
+                    // if(!BrandIds.Contains(trade.ProductType)) SycnSpTradeService.Instance.DoQueueTrade(db, pdb, trade);
+                    // if(trade.ProductType == "23") SycnSpTradeWifiService.Instance.DoQueueTrade(db, trade);
+                }
+                else
+                {
+                    db.Dispose();
+                    pdb.Dispose();
+                    conn = false;
+                    Thread.Sleep(5000);
+                }
+            }
+            catch(Exception ex)
+            {
+                Utils.WriteLog(ex.ToString(), "SP原始数据队列异常");
+            }
+        }
+    }
+
+}

AppStart/Helper/SycnSpServer/SycnSpTradeService.cs

@@ -34,6 +34,7 @@ namespace MySystem
                     BrandIds.Add("24");
                     BrandIds.Add("25");
                     BrandIds.Add("26");
+                    BrandIds.Add("32");
                     WebCMSEntities spdb = new WebCMSEntities();
                     PxcModels.WebCMSEntities db = new PxcModels.WebCMSEntities();
                     JavaProductModels.WebCMSEntities pdb = new JavaProductModels.WebCMSEntities();
@@ -43,137 +44,13 @@ namespace MySystem
                     IQueryable<TradeRecord> trades = spdb.TradeRecord.Where(m => m.Id >= StartId && !BrandIds.Contains(m.ProductType) && m.CreateDate >= start && m.Status == 1).OrderBy(m => m.CreateDate).Take(20);
                     foreach (TradeRecord trade in trades.ToList())
                     {
-                        string TradeSnNo = trade.TradeSnNo;
-                        if(trade.ProductType == "30")
-                        {
-                            trade.MerNo = trade.Field2;
-                        }
-
-                        bool op = true;
-                        //判断盒易付吱码交易
-                        if(trade.ProductType == "12" && trade.TradeType == "2")
-                        {
-                            JavaProductModels.KxsMerchant merchant = pdb.KxsMerchant.FirstOrDefault(m => m.MerchantNo == trade.MerNo);
-                            if(merchant != null)
-                            {
-                                JavaProductModels.KxsMachine pos = pdb.KxsMachine.FirstOrDefault(m => m.Id == merchant.MachineId);
-                                if(pos != null)
-                                {
-                                    trade.TradeSnNo = pos.PosSn;
-                                }
-                            }
-                        }
-                        if (trade.SerEntryMode == "N" && trade.ProductType == "1") //金控押金获取
-                        {
-                            if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
-                            {
-                                op = false;
-                                PosPushDataNewHelper.Deposit(trade);
-                            }
-                        }
-                        else if (trade.SerEntryMode == "1" && trade.ProductType == "27") //金控押金获取
-                        {
-                            if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
-                            {
-                                op = false;
-                                PosPushDataNewHelper.Deposit(trade);
-                            }
-                        }
-                        else if ((trade.ProductType == "4" || trade.ProductType == "8" || trade.ProductType == "9") && (trade.ErrorCode == "5200" || trade.ErrorCode == "5800")) //乐刷海科费率0.52,0.58的不入库
-                        {
-                            op = false;
-                        }
-                        else if (trade.SerEntryMode == "1" && trade.ProductType == "10") //联动押金获取
-                        {
-                            if (trade.TradeAmount == 99 || trade.TradeAmount == 199 || trade.TradeAmount == 299)
-                            {
-                                op = false;
-                                PosPushDataNewHelper.Deposit(trade);
-                            }
-                        }
-                        else if (trade.SerEntryMode == "40" && trade.ProductType == "12") //盒易付押金获取
-                        {
-                            if (trade.TradeAmount == 99 || trade.TradeAmount == 199 || trade.TradeAmount == 299)
-                            {
-                                op = false;
-                                PosPushDataNewHelper.Deposit(trade);
-                                //创业帮押金数据推送
-                                string conn = ConfigurationManager.AppSettings["PushSqlConnStr"].ToString();
-                                CustomerSqlConn.op("INSERT INTO `u_kxs_trade_info_" + trade.CreateDate.Value.ToString("yyyyMM") + "` (`user_id`, `cash_flag`, `merch_no`, `agent_id`, `pos_sn`, `order_id`, `amount`, `trans_time`, `card_type`, `pay_type`) VALUES (123005, 1, '" + trade.MerNo + "', '01', '" + trade.TradeSnNo + "', '" + trade.TradeSerialNo + "', " + trade.TradeAmount + ", '" + trade.CreateDate.Value.ToString("yyyy-MM-dd HH:mm:ss") + "', 1, 0);", conn);
-                            }
-                        }
-                        else if (trade.SettleFee == 0 && trade.ProductType == "15") //联客宝押金获取
-                        {
-                            if (trade.TradeAmount == 99 || trade.TradeAmount == 199 || trade.TradeAmount == 299)
-                            {
-                                op = false;
-                            }
-                        }
-                        else if (trade.Field2 == "1590" && trade.ProductType == "22") //立刷微智能押金获取
-                        {
-                            if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
-                            {
-                                op = false;
-                            }
-                        }
-                        else if (trade.SerEntryMode == "DEPOSIT" && trade.ProductType == "30") //拉卡拉押金获取
-                        {
-                            if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
-                            {
-                                op = false;
-                                if(trade.ProductType == "30")
-                                {
-                                    PxcModels.LklMerNo mer = db.LklMerNo.FirstOrDefault(m => m.ExtMerNo == trade.MerNo);
-                                    if(mer != null)
-                                    {
-                                        trade.MerNo = mer.MerNo;
-                                    }
-                                }
-                                PosPushDataNewHelper.Deposit(trade);
-                            }
-                        }
-                        else if (trade.SerEntryMode == "MACHINE" && trade.ProductType == "34") //联动掌中宝押金获取
-                        {
-                            PosPushDataNewHelper.Deposit(trade);
-                            op = false;
-                        }
-                        if (trade.DiscountRateFlag == "True")
-                        {
-                            op = false;
-                        }
-                        if(trade.ProductType == "7" && trade.TradeStatus != "00")
-                        {
-                            op = false;
-                        }
-                        if (op)
-                        {
-                            bool check = db.SpOrderNos.Any(m => m.OrderNo == trade.TradeSerialNo);
-                            if (!check)
-                            {
-                                PosPushDataNewHelper.Trade(trade);
-                                //创业帮交易数据推送
-                                if(trade.ProductType == "12")
-                                {
-                                    int BankCardType = 1;
-                                    if (trade.BankCardType == "C")
-                                    {
-                                        BankCardType = 1;
-                                    }
-                                    else if (trade.BankCardType == "D")
-                                    {
-                                        BankCardType = 0;
-                                    }
-                                    string conn = ConfigurationManager.AppSettings["PushSqlConnStr"].ToString();
-                                    CustomerSqlConn.op("INSERT INTO `u_kxs_trade_info_" + trade.CreateDate.Value.ToString("yyyyMM") + "` (`user_id`, `cash_flag`, `merch_no`, `agent_id`, `pos_sn`, `order_id`, `amount`, `trans_time`, `card_type`, `pay_type`) VALUES (123005, 0, '" + trade.MerNo + "', '01', '" + trade.TradeSnNo + "', '" + trade.TradeSerialNo + "', " + trade.TradeAmount + ", '" + trade.CreateDate.Value.ToString("yyyy-MM-dd HH:mm:ss") + "', " + BankCardType + ", 0);", conn);
-                                }
-                            }
-                        }
+                        DoQueueTrade(db, pdb, trade);
                         TradeRecord edit = spdb.TradeRecord.FirstOrDefault(m => m.Id == trade.Id);
                         if (edit != null)
                         {
                             edit.Status = 2;
                         }
-                        spdb.SaveChanges();   
+                        spdb.SaveChanges(); 
                     }
                     spdb.Dispose();
                     db.Dispose();
@@ -187,6 +64,135 @@ namespace MySystem
             }
         }
 
+        public void DoQueueTrade(PxcModels.WebCMSEntities db, JavaProductModels.WebCMSEntities pdb, TradeRecord trade)
+        {
+            string TradeSnNo = trade.TradeSnNo;
+            if(trade.ProductType == "30")
+            {
+                trade.MerNo = trade.Field2;
+            }
+
+            bool op = true;
+            //判断盒易付吱码交易
+            if(trade.ProductType == "12" && trade.TradeType == "2")
+            {
+                JavaProductModels.KxsMerchant merchant = pdb.KxsMerchant.FirstOrDefault(m => m.MerchantNo == trade.MerNo);
+                if(merchant != null)
+                {
+                    JavaProductModels.KxsMachine pos = pdb.KxsMachine.FirstOrDefault(m => m.Id == merchant.MachineId);
+                    if(pos != null)
+                    {
+                        trade.TradeSnNo = pos.PosSn;
+                    }
+                }
+            }
+            if (trade.SerEntryMode == "N" && trade.ProductType == "1") //金控押金获取
+            {
+                if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
+                {
+                    op = false;
+                    PosPushDataNewHelper.Deposit(trade);
+                }
+            }
+            else if (trade.SerEntryMode == "1" && trade.ProductType == "27") //金控押金获取
+            {
+                if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
+                {
+                    op = false;
+                    PosPushDataNewHelper.Deposit(trade);
+                }
+            }
+            else if ((trade.ProductType == "4" || trade.ProductType == "8" || trade.ProductType == "9") && (trade.ErrorCode == "5200" || trade.ErrorCode == "5800")) //乐刷海科费率0.52,0.58的不入库
+            {
+                op = false;
+            }
+            else if (trade.SerEntryMode == "1" && trade.ProductType == "10") //联动押金获取
+            {
+                if (trade.TradeAmount == 99 || trade.TradeAmount == 199 || trade.TradeAmount == 299)
+                {
+                    op = false;
+                    PosPushDataNewHelper.Deposit(trade);
+                }
+            }
+            else if (trade.SerEntryMode == "40" && trade.ProductType == "12") //盒易付押金获取
+            {
+                if (trade.TradeAmount == 99 || trade.TradeAmount == 199 || trade.TradeAmount == 299)
+                {
+                    op = false;
+                    PosPushDataNewHelper.Deposit(trade);
+                    //创业帮押金数据推送
+                    string conn = ConfigurationManager.AppSettings["PushSqlConnStr"].ToString();
+                    CustomerSqlConn.op("INSERT INTO `u_kxs_trade_info_" + trade.CreateDate.Value.ToString("yyyyMM") + "` (`user_id`, `cash_flag`, `merch_no`, `agent_id`, `pos_sn`, `order_id`, `amount`, `trans_time`, `card_type`, `pay_type`) VALUES (123005, 1, '" + trade.MerNo + "', '01', '" + trade.TradeSnNo + "', '" + trade.TradeSerialNo + "', " + trade.TradeAmount + ", '" + trade.CreateDate.Value.ToString("yyyy-MM-dd HH:mm:ss") + "', 1, 0);", conn);
+                }
+            }
+            else if (trade.SettleFee == 0 && trade.ProductType == "15") //联客宝押金获取
+            {
+                if (trade.TradeAmount == 99 || trade.TradeAmount == 199 || trade.TradeAmount == 299)
+                {
+                    op = false;
+                }
+            }
+            else if (trade.Field2 == "1590" && trade.ProductType == "22") //立刷微智能押金获取
+            {
+                if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
+                {
+                    op = false;
+                }
+            }
+            else if (trade.SerEntryMode == "DEPOSIT" && trade.ProductType == "30") //拉卡拉押金获取
+            {
+                if (trade.TradeAmount == 9900 || trade.TradeAmount == 19900 || trade.TradeAmount == 29900)
+                {
+                    op = false;
+                    if(trade.ProductType == "30")
+                    {
+                        PxcModels.LklMerNo mer = db.LklMerNo.FirstOrDefault(m => m.ExtMerNo == trade.MerNo);
+                        if(mer != null)
+                        {
+                            trade.MerNo = mer.MerNo;
+                        }
+                    }
+                    PosPushDataNewHelper.Deposit(trade);
+                }
+            }
+            else if (trade.SerEntryMode == "MACHINE" && trade.ProductType == "34") //联动掌中宝押金获取
+            {
+                PosPushDataNewHelper.Deposit(trade);
+                op = false;
+            }
+            if (trade.DiscountRateFlag == "True")
+            {
+                op = false;
+            }
+            if(trade.ProductType == "7" && trade.TradeStatus != "00")
+            {
+                op = false;
+            }
+            if (op)
+            {
+                bool check = db.SpOrderNos.Any(m => m.OrderNo == trade.TradeSerialNo);
+                if (!check)
+                {
+                    PosPushDataNewHelper.Trade(trade);
+                    //创业帮交易数据推送
+                    if(trade.ProductType == "12")
+                    {
+                        int BankCardType = 1;
+                        if (trade.BankCardType == "C")
+                        {
+                            BankCardType = 1;
+                        }
+                        else if (trade.BankCardType == "D")
+                        {
+                            BankCardType = 0;
+                        }
+                        string conn = ConfigurationManager.AppSettings["PushSqlConnStr"].ToString();
+                        CustomerSqlConn.op("INSERT INTO `u_kxs_trade_info_" + trade.CreateDate.Value.ToString("yyyyMM") + "` (`user_id`, `cash_flag`, `merch_no`, `agent_id`, `pos_sn`, `order_id`, `amount`, `trans_time`, `card_type`, `pay_type`) VALUES (123005, 0, '" + trade.MerNo + "', '01', '" + trade.TradeSnNo + "', '" + trade.TradeSerialNo + "', " + trade.TradeAmount + ", '" + trade.CreateDate.Value.ToString("yyyy-MM-dd HH:mm:ss") + "', " + BankCardType + ", 0);", conn);
+                    }
+                }
+            }
+        }
+
         public void StartSim()
         {
             Thread th = new Thread(StartSimDo);

AppStart/Helper/SycnSpServer/SycnSpTradeWifiService.cs

@@ -35,80 +35,7 @@ namespace MySystem
                     IQueryable<TradeRecord> trades = spdb.TradeRecord.Where(m => m.Id >= StartId && m.ProductType == "23" && m.CreateDate >= start && m.Status == 1).OrderBy(m => m.CreateDate).Take(20);
                     foreach (TradeRecord trade in trades.ToList())
                     {
-                        bool check = db.SpOrderNos.Any(m => m.OrderNo == trade.TradeSerialNo);
-                        if (!check && trade.SerEntryMode != "1")
-                        {
-                            int BrandId = int.Parse(function.CheckInt(trade.ProductType));
-                            decimal TradeAmount = trade.TradeAmount;
-                            if(trade.SeoTitle != "v2") TradeAmount = TradeAmount / 100;
-                            PxcModels.WifiTradeRecord add = db.WifiTradeRecord.Add(new PxcModels.WifiTradeRecord()
-                            {
-                                CreateDate = trade.CreateDate,
-                                UpdateDate = trade.UpdateDate,
-                                SnNo = trade.TradeSnNo, //渠道SN号
-                                MerNo = trade.MerNo, //商户号
-                                TradeAmount = TradeAmount, //交易金额
-                                BrandId = BrandId, //品牌
-                                Remark = trade.MerNo, //备注
-                                OrderNo = trade.TradeSerialNo,
-                            }).Entity;
-                            db.SpOrderNos.Add(new PxcModels.SpOrderNos()
-                            {
-                                OrderNo = trade.TradeSerialNo
-                            });
-                            db.SaveChanges();
-
-                            //开始统计
-                            int Months = 0;
-
-                            if(trade.SeoTitle == "v2")
-                            {
-                                string duration = trade.Field2; //设备套餐时长
-                                string unit = trade.Field3; //套餐单位 0:按天 1:按月
-                                if(unit == "1") Months = int.Parse(function.CheckInt(duration));
-                                else Months = int.Parse(function.CheckInt(duration)) / 30;
-                            }
-                            else
-                            {
-                                if(BrandId == 23 && TradeAmount == 49) Months = 1;
-                                else if(BrandId == 23 && TradeAmount == 79) Months = 3;
-                                else if(BrandId == 23 && TradeAmount == 119) Months = 12;
-                                else if(BrandId == 23 && TradeAmount == 199) Months = 24;
-                                else Months = trade.QueryCount;
-
-                                // if(BrandId == 24 && TradeAmount == 59) Months = 1;
-                                // if(BrandId == 24 && TradeAmount == 159) Months = 3;
-                                // if(BrandId == 24 && TradeAmount == 399) Months = 12;
-                                // if(BrandId == 24 && TradeAmount == 699) Months = 24;
-
-                                // if(BrandId == 25 && TradeAmount == 69) Months = 1;
-                                // if(BrandId == 25 && TradeAmount == 199) Months = 3;
-                                // if(BrandId == 25 && TradeAmount == 499) Months = 12;
-                                // if(BrandId == 25 && TradeAmount == 799) Months = 24;
-
-                                // if(BrandId == 26 && TradeAmount == 219) Months = 1;
-                                // if(BrandId == 26 && TradeAmount == 449) Months = 3;
-                                // if(BrandId == 26 && TradeAmount == 1399) Months = 12;
-                                // if(BrandId == 26 && TradeAmount == 2698) Months = 24;
-                            }
-                            string TradeMonth = DateTime.Now.ToString("yyyyMM");
-
-                            add.Duration = Months;
-                            add.Unit = "m";
-                            WifiTradeHelper.Instance.DoOne(db, add.Id, DateTime.Now.ToString("yyyy-MM"), trade.TradeSerialNo, true);
-                            db.SaveChanges();
-                        }
-                        if(trade.SerEntryMode == "1")
-                        {
-                            PosPushDataNewHelper.Deposit(new ActivateRecord()
-                            {
-                                SeoTitle = trade.TradeAmount.ToString(),
-                                ProductType = trade.ProductType,
-                                SnNo = trade.TradeSnNo,
-                                MerNo = trade.TradeSnNo,
-                                Id = trade.Id,
-                            });
-                        }
+                        DoQueueTrade(db, trade);
                         TradeRecord edit = spdb.TradeRecord.FirstOrDefault(m => m.Id == trade.Id);
                         if (edit != null)
                         {
@@ -127,6 +54,86 @@ namespace MySystem
             }
         }
 
+        public void DoQueueTrade(PxcModels.WebCMSEntities db, TradeRecord trade)
+        {
+            bool check = db.SpOrderNos.Any(m => m.OrderNo == trade.TradeSerialNo);
+            if (!check && trade.SerEntryMode != "1")
+            {
+                int BrandId = int.Parse(function.CheckInt(trade.ProductType));
+                decimal TradeAmount = trade.TradeAmount;
+                if(trade.SeoTitle != "v2") TradeAmount = TradeAmount / 100;
+                PxcModels.WifiTradeRecord add = db.WifiTradeRecord.Add(new PxcModels.WifiTradeRecord()
+                {
+                    CreateDate = trade.CreateDate,
+                    UpdateDate = trade.UpdateDate,
+                    SnNo = trade.TradeSnNo, //渠道SN号
+                    MerNo = trade.MerNo, //商户号
+                    TradeAmount = TradeAmount, //交易金额
+                    BrandId = BrandId, //品牌
+                    Remark = trade.MerNo, //备注
+                    OrderNo = trade.TradeSerialNo,
+                }).Entity;
+                db.SpOrderNos.Add(new PxcModels.SpOrderNos()
+                {
+                    OrderNo = trade.TradeSerialNo
+                });
+                db.SaveChanges();
+
+                //开始统计
+                int Months = 0;
+
+                if(trade.SeoTitle == "v2")
+                {
+                    string duration = trade.Field2; //设备套餐时长
+                    string unit = trade.Field3; //套餐单位 0:按天 1:按月
+                    if(unit == "1") Months = int.Parse(function.CheckInt(duration));
+                    else Months = int.Parse(function.CheckInt(duration)) / 30;
+                }
+                else
+                {
+                    if(BrandId == 23 && TradeAmount == 49) Months = 1;
+                    else if(BrandId == 23 && TradeAmount == 79) Months = 3;
+                    else if(BrandId == 23 && TradeAmount == 119) Months = 12;
+                    else if(BrandId == 23 && TradeAmount == 199) Months = 24;
+                    else Months = trade.QueryCount;
+
+                    // if(BrandId == 24 && TradeAmount == 59) Months = 1;
+                    // if(BrandId == 24 && TradeAmount == 159) Months = 3;
+                    // if(BrandId == 24 && TradeAmount == 399) Months = 12;
+                    // if(BrandId == 24 && TradeAmount == 699) Months = 24;
+
+                    // if(BrandId == 25 && TradeAmount == 69) Months = 1;
+                    // if(BrandId == 25 && TradeAmount == 199) Months = 3;
+                    // if(BrandId == 25 && TradeAmount == 499) Months = 12;
+                    // if(BrandId == 25 && TradeAmount == 799) Months = 24;
+
+                    // if(BrandId == 26 && TradeAmount == 219) Months = 1;
+                    // if(BrandId == 26 && TradeAmount == 449) Months = 3;
+                    // if(BrandId == 26 && TradeAmount == 1399) Months = 12;
+                    // if(BrandId == 26 && TradeAmount == 2698) Months = 24;
+                }
+                string TradeMonth = DateTime.Now.ToString("yyyyMM");
+
+                add.Duration = Months;
+                add.Unit = "m";
+                WifiTradeHelper.Instance.DoOne(db, add.Id, DateTime.Now.ToString("yyyy-MM"), trade.TradeSerialNo, true);
+                db.SaveChanges();
+            }
+            if(trade.SerEntryMode == "1")
+            {
+                PosPushDataNewHelper.Deposit(new ActivateRecord()
+                {
+                    SeoTitle = trade.TradeAmount.ToString(),
+                    ProductType = trade.ProductType,
+                    SnNo = trade.TradeSnNo,
+                    MerNo = trade.TradeSnNo,
+                    Id = trade.Id,
+                });
+            }
+        }
+
+        
+
         public void StatTrade(PxcModels.WebCMSEntities db, int UserId, int BrandId, string TradeMonth, decimal TradeAmount)
         {
             PxcModels.Users user = db.Users.FirstOrDefault(m => m.Id == UserId) ?? new PxcModels.Users();

AppStart/RabbitMQClient2.cs

@@ -0,0 +1,100 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using Library;
+using System.Threading;
+
+namespace MySystem
+{
+    public class RabbitMQClient2
+    {
+        public readonly static RabbitMQClient2 Instance = new RabbitMQClient2();
+        string UserName,Password,HostName,VirtualHostName;
+        private RabbitMQClient2()
+        {
+            UserName = ConfigurationManager.AppSettings["MqUserName"].ToString();
+            Password = ConfigurationManager.AppSettings["MqPassword"].ToString();
+            HostName = ConfigurationManager.AppSettings["MqHostName"].ToString();
+            VirtualHostName = ConfigurationManager.AppSettings["MqVirtualHostName"].ToString();
+        }
+
+        public static IConnection _connection;
+        public void CreateConn()
+        { 
+            var factory = new ConnectionFactory()
+            {
+                HostName = HostName,
+                UserName = UserName,
+                Password = Password,
+                VirtualHost = VirtualHostName
+            };
+            _connection = factory.CreateConnection();
+        }
+
+        #region 单对单接收
+
+        public void StartReceive(string QueueName, Action<string> CallBack)
+        {
+            if (_connection == null)
+            {
+                CreateConn();
+            }
+            else if (!_connection.IsOpen)
+            {
+                CreateConn();
+            }
+
+            var consumer = new EventingBasicConsumer(_channel[QueueName]);
+            consumer.Received += (model, ea) =>
+            {
+                var body = ea.Body.ToArray();
+                //获取接收的数据
+                var message = Encoding.UTF8.GetString(body);
+
+                // 模拟消息处理逻辑
+                try
+                {
+                    CallBack(message);
+                    // 手动确认消息
+                    _channel[QueueName].BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
+                }
+                catch (Exception ex)
+                {
+                    // 如果处理失败，可以选择拒绝消息或重新入队
+                    _channel[QueueName].BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true);
+                    Utils.WriteLog(ex.ToString(), "MQ异常");
+                }
+            };
+
+            // 设置 autoAck = false
+            _channel[QueueName].BasicConsume(queue: QueueName, autoAck: false, consumer: consumer);
+        }
+        #endregion
+
+        #region 单对单发送
+        public Dictionary<string, IModel> _channel = new Dictionary<string, IModel>();
+        public void Conn(string QueueName)
+        {
+            if (_connection == null)
+            {
+                CreateConn();
+            }
+            else if (!_connection.IsOpen)
+            {
+                CreateConn();
+            }
+            var channel = _connection.CreateModel();
+            channel.ExchangeDeclare("kxs_direct_ranch", "direct", true);
+            channel.QueueDeclare(QueueName, true, false, false);
+            channel.QueueBind(QueueName, "kxs_direct_ranch", QueueName);
+            if(!_channel.ContainsKey(QueueName)) _channel.Add(QueueName, channel);
+        }
+        public void Push(string QueueName, string Content)
+        {
+            _channel[QueueName].BasicPublish("", QueueName, null, Encoding.Default.GetBytes(Content));
+        }
+        #endregion
+    }
+}

AppStart/RabbitMQClientV2.cs

@@ -1,121 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-using RabbitMQ.Client;
-using RabbitMQ.Client.Events;
-using Library;
-using System.Threading;
-using LitJson;
-
-namespace MySystem
-{
-    public class RabbitMQClientV2
-    {
-        public readonly static RabbitMQClientV2 Instance = new RabbitMQClientV2();
-        string UserName,Password,HostName,VirtualHostName;
-        private RabbitMQClientV2()
-        {
-            UserName = ConfigurationManager.AppSettings["MqUserName"].ToString();
-            Password = ConfigurationManager.AppSettings["MqPassword"].ToString();
-            HostName = ConfigurationManager.AppSettings["MqHostName"].ToString();
-            VirtualHostName = ConfigurationManager.AppSettings["MqVirtualHostName"].ToString();
-        }
-
-        #region 单对单发送
-        public void SendMsg(string content, string QueueName, uint delayMilliseconds = 0)
-        {
-            if(delayMilliseconds > 0)
-            {
-                // 设置消息的延迟时间
-                var properties = _channel_send.CreateBasicProperties();
-                properties.Headers = new Dictionary<string, object>
-                {
-                    { "x-delay", delayMilliseconds } // 延迟5秒
-                };
-                _channel_send.BasicPublish("kxs_delay_ranch", QueueName, properties, Encoding.Default.GetBytes(content));
-            }
-            else
-            {
-                _channel_send.BasicPublish("", QueueName, null, Encoding.Default.GetBytes(content));
-            }
-            Utils.WriteLog(DateTime.Now.ToString() + "\n" + QueueName + "\n" + content + "\n\n\n", "SendMsg2");
-        }
-        #endregion
-
-        #region 单对单接收
-        public static IModel _channel_send;
-        public void CreateConn(string QueueName)
-        { 
-            var factory = new ConnectionFactory()
-            {
-                UserName = UserName,
-                Password = Password,
-                AutomaticRecoveryEnabled = true,  //如果connection挂掉是否重新连接
-                TopologyRecoveryEnabled = true,  //连接恢复后，连接的交换机，队列等是否一同恢复
-                VirtualHost = VirtualHostName,
-            };
-            List<AmqpTcpEndpoint> p = new List<AmqpTcpEndpoint>();
-            string[] HostNames = HostName.Split(',');
-            foreach (string subHostName in HostNames)
-            {
-                string[] subHostNameData = subHostName.Split(':');
-                p.Add(new AmqpTcpEndpoint(subHostNameData[0], int.Parse(subHostNameData[1])));
-            }
-            _connection = factory.CreateConnection(p);
-            _channel_send = _connection.CreateModel();
-            _channel_send.ExchangeDeclare("kxs_delay_ranch", "x-delayed-message", true);
-            _channel_send.QueueDeclare(QueueName, true, false, false);
-            string routingKey = "delayed_routing_key";
-            IDictionary<string, object> arguments = new Dictionary<string, object>
-            {
-                { "x-delayed-type", "direct" } // 这里可以指定不同的交换机类型
-            };
-            _channel_send.QueueBind(QueueName, "kxs_delay_ranch", routingKey, arguments);
-        }
-
-        public static IConnection _connection;
-        public void CreateConn()
-        { 
-            var factory = new ConnectionFactory()
-            {
-                UserName = UserName,
-                Password = Password,
-                AutomaticRecoveryEnabled = true,  //如果connection挂掉是否重新连接
-                TopologyRecoveryEnabled = true,  //连接恢复后，连接的交换机，队列等是否一同恢复
-                RequestedHeartbeat = TimeSpan.FromMinutes(1),
-                VirtualHost = VirtualHostName,
-            };
-            List<AmqpTcpEndpoint> p = new List<AmqpTcpEndpoint>();
-            string[] HostNames = HostName.Split(',');
-            foreach (string subHostName in HostNames)
-            {
-                string[] subHostNameData = subHostName.Split(':');
-                p.Add(new AmqpTcpEndpoint(subHostNameData[0], int.Parse(subHostNameData[1])));
-            }
-            _connection = factory.CreateConnection(p);
-        }
-        public void StartReceive(string QueueName)
-        {
-            if (_connection == null)
-            {
-                CreateConn(QueueName);
-            }
-            else if (!_connection.IsOpen)
-            {
-                CreateConn(QueueName);
-            }
-            var channel = _connection.CreateModel();
-            channel.QueueBind(QueueName, "kxs_delay_ranch", QueueName);
-            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
-            consumer.Received += (a, e) =>
-            {
-                string MsgContent = Encoding.Default.GetString(e.Body.ToArray());
-                SycnSpTradeWifiService.Instance.PushMsgToJava(MsgContent);
-                channel.BasicAck(e.DeliveryTag, true); //收到回复后，RabbitMQ会直接在队列中删除这条消息
-            };
-            channel.BasicConsume(QueueName, false, consumer);
-        }
-        #endregion
-
-    }
-}

Startup.cs

@@ -190,6 +190,11 @@ namespace MySystem
                 WifiChangeBindHelper.Instance.Start(); //wifi换绑
                 SycnJavaUsersService.Instance.Start(); //同步创客基本信息
 
+
+                RabbitMQClient2.Instance.Conn("MainDataQueue");
+                SpDataHelper.Instance.Start("MainDataQueue");
+                SpDataHelper.Instance.StartData();
+
                 // WifiCheckHelper.Instance.DoSomething();
 
                 // StatService.Instance.StartActiveReward(); //实时处理激活奖励