kxs-main/AppStart/Helper/SpDataHelper.cs

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(Newtonsoft.Json.JsonConvert.SerializeObject(trade), "SP原始数据队列日志");
                    // if(!BrandIds.Contains(trade.ProductType)) SycnSpTradeService.Instance.DoQueueTrade(db, pdb, trade);
                    // if(trade.ProductType == "23") SycnSpTradeWifiService.Instance.DoQueueTrade(db, trade);
                }
                else
                {
                    if(conn)
                    {
                        db.Dispose();
                        pdb.Dispose();
                        conn = false;
                    }
                    Thread.Sleep(5000);
                }
            }
            catch(Exception ex)
            {
                Utils.WriteLog(ex.ToString(), "SP原始数据队列异常");
            }
        }
    }

}