添加昌乐的数据同步

TimedUpload/Constants.cs

@@ -28,6 +28,7 @@ namespace TimedUpload
         public static string ChargeDB = ConfigurationManager.AppSettings["ChargeDB"];
         public static string SecondaryToDMA = ConfigurationManager.AppSettings["SecondaryToDMA"];
         public static string SecondaryToDMACode = ConfigurationManager.AppSettings["SecondaryToDMACode"];
+        public static string changleWaterFactoryDb = ConfigurationManager.AppSettings["changleWaterFactoryDb"];
 
     }
 }

TimedUpload/QuartzJobs/ChangleBusinessDataJob.cs

@@ -0,0 +1,238 @@
+using log4net;
+using Quartz;
+using RabbitMQ.Client;
+using RDIFramework.Utilities;
+using System;
+using System.Collections.Generic;
+using System.Configuration;
+using System.Data;
+using System.Linq;
+using System.Text;
+
+namespace TimedUpload.QuartzJobs
+{
+    [DisallowConcurrentExecution]
+    public class ChangleBusinessDataJob : IJob
+    {
+        private readonly ILog log = LogManager.GetLogger(typeof(DABusinessDataJob));
+        private string manufacturerCode = Constants.ManufacturerCode;
+        private string bmId = Constants.BmId;
+        private string readDate = Constants.UserMeterDate;
+        private string userMeterReadDate = Constants.UserMeterReadDate;
+        public void Execute(IJobExecutionContext context)
+        {
+            string[] uploadUrls = Constants.UploadUrl.Split('|');
+            Dictionary<string, IConnection> connections = new Dictionary<string, IConnection>();
+            Dictionary<string, IModel> channels = new Dictionary<string, IModel>();
+            Dictionary<string, IBasicProperties> properties = new Dictionary<string, IBasicProperties>();
+
+            foreach (string uploadUrl in uploadUrls)
+            {
+                ConnectionFactory factory = new ConnectionFactory();
+                factory.HostName = uploadUrl;//主机名，Rabbit会拿这个IP生成一个endpoint，这个很熟悉吧，就是socket绑定的那个终结点。
+                factory.UserName = Constants.UploadUserName;//默认用户名,用户可以在服务端自定义创建，有相关命令行
+                factory.Password = Constants.UploadPassword;//默认密码
+
+                factory.AutomaticRecoveryEnabled = true; // 链接断开会自动重连
+
+                IConnection connection = factory.CreateConnection();
+                IModel channel = connection.CreateModel();
+                channel.QueueDeclare("zone.userMeter", true, false, false, null);//创建一个名称为kibaqueue的消息队列
+                channel.QueueDeclare("zone.userMeterHis", true, false, false, null);//创建一个名称为kibaqueue的消息队列
+
+                IBasicProperties property = channel.CreateBasicProperties();
+                property.ContentType = "text/plain";
+                property.DeliveryMode = 2; //持久化
+                connections.Add(uploadUrl, connection);
+                properties.Add(uploadUrl, property);
+                channels.Add(uploadUrl, channel);
+            }
+
+            if (channels.Count > 0)
+            {
+                SendZoneMeterUser(channels, properties);
+                SendZoneMeterUserHis(channels, properties);
+            }
+
+            foreach (KeyValuePair<string, IConnection> item in connections)
+            {
+                IConnection connection = item.Value;
+                connection.Close();
+            }
+
+        }
+
+        /// <summary>
+        /// 户表添加
+        /// </summary>
+        /// <param name="channel"></param>
+        private void SendZoneMeterUser(Dictionary<string, IModel> channels, Dictionary<string, IBasicProperties> properties)
+        {
+            string meterId = Constants.UserMeterId;
+            log.Info("营收户表基础数据同步任务开始执行.................\r\n");
+            while (true)
+            {
+                DataTable dt = null;
+                try
+                {
+                    string sql = "SELECT top 100 b.CM_ID,c.RouteCode,RouteName,a.CustomerCode,a.CustomerName,b.LifetimeCode ElecAddress,b.DetailedAddress MeterAddress,a.DetailedAddress CustomerAddress";
+                    sql += " FROM BCS_Customer a,BCS_CustomerMeter b, BCS_MeterReadingRoute c WHERE a.Cus_ID = b.Cus_ID and b.Mrr_ID = c.MRR_ID ";
+                    sql += " AND b.CM_ID > " + meterId + " ORDER BY CM_ID";
+
+                    dt = dbHelper.Fill(sql);
+                }
+                catch (Exception ex)
+                {
+                    log.Info("营收户表基础数据同步查询数据异常" + ex.StackTrace + "\r\n");
+                }
+
+                if (dt == null || dt.Rows.Count == 0)
+                {
+                    break;
+                }
+
+                log.Info("营收户表基础数据同步获取记录数：【" + dt.Rows.Count + "】................\r\n");
+
+                StringBuilder message = new StringBuilder();
+                for (int i = 0; i < dt.Rows.Count; i++)
+                {
+                    message.Clear();
+                    try
+                    {
+                        DataRow dr = dt.Rows[i];
+
+                        meterId = Convert.ToString(dr["CM_ID"]);
+
+                        string meterLineNo = Convert.ToString(dr["RouteCode"]);
+                        string meterLineName = Convert.ToString(dr["RouteName"]);
+                        string clientNo = Convert.ToString(dr["CustomerCode"]);
+                        string clientName = Convert.ToString(dr["CustomerName"]);
+                        string meterCode = Convert.ToString(dr["CM_ID"]);
+                        string meterAddress = string.IsNullOrEmpty(Convert.ToString(dr["CustomerAddress"])) ? "未知" : Convert.ToString(dr["CustomerAddress"]);
+
+                        message.Append("{");
+                        message.Append("\"meterLineNo\": \"").Append(meterLineNo).Append("\",");
+                        message.Append("\"meterLineName\": \"").Append(meterLineName).Append("\",");
+                        message.Append("\"clientNo\": \"").Append(clientNo).Append("\",");
+                        message.Append("\"clientName\": \"").Append(clientName).Append("\",");
+                        message.Append("\"meterCode\": \"").Append(meterCode).Append("\",");
+                        message.Append("\"meterAddress\": \"").Append(meterAddress).Append("\",");
+                        message.Append("\"fromWhere\": \"").Append(manufacturerCode).Append("\"");
+                        message.Append("}");
+
+                        foreach (KeyValuePair<string, IModel> item in channels)
+                        {
+                            string key = item.Key;
+                            IModel channel = item.Value;
+                            IBasicProperties property = properties[key];
+                            channel.BasicPublish("zone.userMeter", "", property, Encoding.UTF8.GetBytes(message.ToString())); //生产消息
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        log.Info("营收户表基础数据推送失败：" + message.ToString() + "\r\n");
+                        log.Error(ex.Message + "===========" + ex.StackTrace + "\r\n");
+                    }
+                }
+
+            }
+            UpdateAppConfig("UserMeterId", meterId);
+            log.Info("营收户表基础数据同步任务结束执行.................\r\n");
+
+        }
+
+        /// <summary>
+        /// 户表历史数据
+        /// </summary>
+        /// <param name="channel"></param>
+        private void SendZoneMeterUserHis(Dictionary<string, IModel> channels, Dictionary<string, IBasicProperties> properties)
+        {
+            string userMeterReadId = Constants.UserMeterReadId;
+            log.Info("营收户表抄表数据同步任务开始执行.................\r\n");
+            while (true)
+            {
+                DataTable dt = null;
+                try
+                {
+                    string sql = "SELECT top 100 UsedWater_ID,CM_ID,ThisMeterNumber,ThisMeterDt,BetweenMeteNumber,b.CreateDT FROM BCS_UsedWater b";
+                    sql += " WHERE b.BM_ID >= " + bmId + " AND b.UsedWater_ID > " + userMeterReadId + " ORDER BY b.BM_ID,b.UsedWater_ID";
+                    dt = dbHelper.Fill(sql);
+                }
+                catch (Exception ex)
+                {
+                    log.Info("营收户表抄表数据同步查询数据异常" + ex.StackTrace + "\r\n");
+                }
+
+                if (dt == null || dt.Rows.Count == 0)
+                {
+
+                    break;
+                }
+                log.Info("营收户表抄表数据同步获取记录数：【" + dt.Rows.Count + "】................\r\n");
+
+                StringBuilder message = new StringBuilder();
+                for (int i = 0; i < dt.Rows.Count; i++)
+                {
+                    message.Clear();
+                    try
+                    {
+                        DataRow dr = dt.Rows[i];
+                        userMeterReadId = Convert.ToString(dr["UsedWater_ID"]);
+
+                        string meterCode = Convert.ToString(dr["CM_ID"]);
+                        string getDateTime = Convert.ToDateTime(dr["ThisMeterDt"]).ToString("yyyy-MM-dd HH:mm:ss");
+                        string currReadingValue = Convert.ToString(dr["ThisMeterNumber"]);
+                        string waterUsed = Convert.ToString(dr["BetweenMeteNumber"]);
+
+                        message.Append("{");
+                        message.Append("\"meterCode\": \"").Append(meterCode).Append("\",");
+                        message.Append("\"fromWhere\": \"").Append(manufacturerCode).Append("\",");
+                        message.Append("\"saleWater\": \"").Append(waterUsed).Append("\",");
+                        message.Append("\"getDateTime\": \"").Append(getDateTime).Append("\",");
+                        message.Append("\"currReadingValue\": ").Append(currReadingValue);
+                        message.Append("}");
+
+                        foreach (KeyValuePair<string, IModel> item in channels)
+                        {
+                            string key = item.Key;
+                            IModel channel = item.Value;
+                            IBasicProperties property = properties[key];
+                            channel.BasicPublish("zone.userMeterHis", "", property, Encoding.UTF8.GetBytes(message.ToString())); //生产消息
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        log.Info("营收户表抄表数据推送失败：" + message.ToString() + "\r\n");
+                        log.Error(ex.Message + "===========" + ex.StackTrace + "\r\n");
+                    }
+                }
+            }
+            UpdateAppConfig("UserMeterReadId", userMeterReadId);
+            log.Info("营收户表抄表数据同步任务结束执行.................\r\n");
+
+        }
+
+        /// <summary>
+        /// 更新配置文件中的值
+        /// </summary>
+        /// <param name="key">键</param>
+        /// <param name="value">值</param>
+        private void UpdateAppConfig(String key, String value)
+        {
+            var cfg = ConfigurationManager.OpenExeConfiguration(ConfigurationUserLevel.None);
+            cfg.AppSettings.Settings[key].Value = value;
+            cfg.Save();
+            ConfigurationManager.RefreshSection("appSettings");
+
+        }
+
+        static IDbProvider dbHelper
+        {
+            get
+            {
+                var DbDefine = DbFactoryProvider.GetProvider(CurrentDbType.SqlServer, Constants.ChargeDB);
+                return DbDefine;
+            }
+        }
+    }
+}

TimedUpload/QuartzJobs/ChangleSecondPumpDataJob.cs

@@ -89,7 +89,7 @@ namespace TimedUpload.QuartzJobs
 
                 for (int k = 0; k < dtDevice.Rows.Count; k++)
                 {
-                    string table = "历史记录_" + dtDevice.Rows[k]["devId"].ToString().PadLeft(6,'0')+DateTime.Now.Year.ToString();
+                    string table = "历史记录_" + dtDevice.Rows[k]["devId"].ToString().PadLeft(6,'0')+"_"+DateTime.Now.Year.ToString();
                     String sql = @"SELECT TOP 2000 [id]
                         ,[设备ID],[记录时间],[采集时间]
                         ,[设备状态],[通讯状态],[数据来源]

TimedUpload/QuartzJobs/ChangleWaterFactoryDataJob.cs

@@ -0,0 +1,336 @@
+using log4net;
+using Newtonsoft.Json;
+using Quartz;
+using RabbitMQ.Client;
+using RDIFramework.Utilities;
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Text;
+
+namespace TimedUpload.QuartzJobs
+{
+    [DisallowConcurrentExecution]
+    public class ChangleWaterFactoryDataJob : IJob
+    {
+        private readonly ILog log = LogManager.GetLogger(typeof(ChangleWaterFactoryDataJob));
+
+        public void Execute(IJobExecutionContext context)
+        {
+            string[] uploadUrls = Constants.UploadUrl.Split('|');
+            Dictionary<string, IConnection> connections = new Dictionary<string, IConnection>();
+            Dictionary<string, IModel> channels = new Dictionary<string, IModel>();
+            Dictionary<string, IBasicProperties> properties = new Dictionary<string, IBasicProperties>();
+
+            foreach (string uploadUrl in uploadUrls)
+            {
+                ConnectionFactory factory = new ConnectionFactory();
+                factory.HostName = uploadUrl;//主机名，Rabbit会拿这个IP生成一个endpoint，这个很熟悉吧，就是socket绑定的那个终结点。
+                factory.UserName = Constants.UploadUserName;//默认用户名,用户可以在服务端自定义创建，有相关命令行
+                factory.Password = Constants.UploadPassword;//默认密码
+
+                factory.AutomaticRecoveryEnabled = true; // 链接断开会自动重连
+
+                IConnection connection = factory.CreateConnection();
+                IModel channel = connection.CreateModel();
+                channel.QueueDeclare("waterFactory.deviceHis", true, false, false, null);//创建一个名称为kibaqueue的消息队列
+
+                IBasicProperties property = channel.CreateBasicProperties();
+                property.ContentType = "text/plain";
+                property.DeliveryMode = 2; //持久化
+                connections.Add(uploadUrl, connection);
+                properties.Add(uploadUrl, property);
+                channels.Add(uploadUrl, channel);
+            }
+
+            if (channels.Count > 0)
+            {
+                SendWaterFactoryHis(channels, properties);
+            }
+
+            foreach (KeyValuePair<string, IConnection> item in connections)
+            {
+                IConnection connection = item.Value;
+                connection.Close();
+            }
+        }
+
+        /// <summary>
+        /// 水厂历史数据
+        /// </summary>
+        /// <param name="channels"></param>
+        /// <param name="properties"></param>
+        private void SendWaterFactoryHis(Dictionary<string, IModel> channels, Dictionary<string, IBasicProperties> properties)
+        {
+            try
+            {
+                Object inPh= null, inChlorine = null, inTurbidty = null, inTurbidtytwo = null, outPh = null, outChlorine = null, outTurbidty = null, level = null;
+
+                DateTime nowTime = DateTime.Now;
+                log.Info("昌乐水厂数据同步任务开始执行.................\r\n");
+                string factorySql = "SELECT [id],[inMeterId],[outMeterId],[uploadTime],[waterFactory]  FROM [dbo].[uploadList]";
+                DataTable dtFactory = waterFHelper.Fill(factorySql);
+                if (dtFactory == null || dtFactory.Rows.Count == 0)
+                {
+                    return;
+                }
+
+                // 处理水厂水表
+                for (int i = 0; i < dtFactory.Rows.Count; i++)
+                {
+                    string factoryCode = dtFactory.Rows[i]["waterFactory"].ToString();
+
+                    // DataTable dtMeter = dbHelper.Fill(meterSql);
+
+                    //if (dtMeter == null || dtMeter.Rows.Count == 0)
+                    //{
+                    //    continue;
+                    //}
+                    Dictionary<string, object> factoryMap = new Dictionary<string, object>();
+                    factoryMap["FactoryCode"] = factoryCode;
+
+                    // 处理水厂基础数据
+                    // string baseSql = "SELECT RTRIM(日期) + ' ' +  RTRIM(时间) ReadTime,反冲正累计流量 - 反冲负累计流量 BackWashing,[清水池液位] LiquidHeight,更新时间 FROM [dbo].[水厂泵房2] WHERE FactoryId = " + factoryId + " ORDER BY 日期, 时间 ";
+                    // DataTable baseDt = dbHelper.Fill(baseSql);
+                    DateTime searchTime = Convert.ToDateTime(dtFactory.Rows[i]["uploadTime"]);
+                    TimeSpan sTs = new TimeSpan(searchTime.Ticks);
+                    TimeSpan eTs = new TimeSpan(nowTime.Ticks);
+                    TimeSpan ts = eTs - sTs;
+                    int num = (int)ts.TotalMinutes / 5;
+                    if (num > 200) {
+                        num = 200;
+                    }
+                    for (int l = 0; l < num; l++)
+                    {
+                        log.Info(l);
+                        // DataRow baseRow = baseDt.Rows[l];
+                        Dictionary<string, object> baseMap = new Dictionary<string, object>();
+                        ArrayList meterDataList = new ArrayList();
+                        searchTime = searchTime.AddMinutes(5);
+                        string time = searchTime.ToString("yyyy/MM/dd HH:mm:ss");// baseRow["ReadTime"].ToString();
+                        string year = searchTime.Year.ToString();
+                        baseMap["ReadTime"] = time;
+                        baseMap["BackWashing"] = "0";//baseRow["BackWashing"].ToString();
+
+                        baseMap["Consumption"] = "0";
+                        baseMap["CleanWaterFlow"] = "0";
+                        baseMap["Dosage"] = "0";
+
+
+                        // 进水
+                        string insql = @"SELECT TOP 1 [id],[设备ID],[记录时间],[采集时间],[设备状态],[通讯状态],[净累计流量],[瞬时流量] FROM [dbo].[历史记录_000523_" + year + "] where 采集时间 >= '" + time + "'";
+                        DataTable inDt = dbHelper.Fill(insql);
+                        if (inDt != null && inDt.Rows.Count > 0)
+                        {
+                            Dictionary<string, object> meterMap = new Dictionary<string, object>();
+                            meterMap["InstantaneousFlow"] = inDt.Rows[0]["瞬时流量"];
+                            meterMap["NetCumulativeFlow"] = inDt.Rows[0]["净累计流量"];
+                            meterMap["Pressure"] = null;
+                            meterMap["PH"] = null;
+                            meterMap["Chlorine"] = null;
+                            meterMap["Turbidity"] = null;
+                            meterMap["TurbidityTwo"] = null;
+                            meterMap["ReadTime"] = time;
+                            meterMap["MeterCode"] = "wwkj001";
+                            meterDataList.Add(meterMap);
+                        }
+                        else
+                        {
+                            Dictionary<string, object> meterMap = new Dictionary<string, object>();
+                            meterMap["InstantaneousFlow"] = null;
+                            meterMap["NetCumulativeFlow"] = null;
+                            meterMap["Pressure"] = null;
+                            meterMap["PH"] = null;
+                            meterMap["Chlorine"] = null;
+                            meterMap["Turbidity"] = null;
+                            meterMap["TurbidityTwo"] = null;
+                            meterMap["ReadTime"] = time;
+                            meterMap["MeterCode"] = "wwkj001";
+                            meterDataList.Add(meterMap);
+                        }
+                        // 出水
+                        string outsql = @"SELECT TOP 1 [id],[设备ID],[记录时间],[采集时间],[设备状态],[通讯状态],[净累计流量],[瞬时流量] FROM [dbo].[历史记录_000286_" + year + "] where 采集时间 >= '" + time + "'"; ;
+                        DataTable outDt = dbHelper.Fill(outsql);
+                        if (outDt != null && outDt.Rows.Count > 0)
+                        {
+                            Dictionary<string, object> meterMap = new Dictionary<string, object>();
+                            meterMap["InstantaneousFlow"] = outDt.Rows[0]["瞬时流量"];
+                            meterMap["NetCumulativeFlow"] = outDt.Rows[0]["净累计流量"];
+                            meterMap["Pressure"] = null;
+                            meterMap["PH"] = null;
+                            meterMap["Chlorine"] = null;
+                            meterMap["Turbidity"] = null;
+                            meterMap["TurbidityTwo"] = null;
+                            meterMap["ReadTime"] = time;
+                            meterMap["MeterCode"] = "wwkj005";
+                            meterDataList.Add(meterMap);
+                        }
+                        else
+                        {
+                            Dictionary<string, object> meterMap = new Dictionary<string, object>();
+                            meterMap["InstantaneousFlow"] = null;
+                            meterMap["NetCumulativeFlow"] = null;
+                            meterMap["Pressure"] = null;
+                            meterMap["PH"] = null;
+                            meterMap["Chlorine"] = null;
+                            meterMap["Turbidity"] = null;
+                            meterMap["TurbidityTwo"] = null;
+                            meterMap["ReadTime"] = time;
+                            meterMap["MeterCode"] = "wwkj005";
+                            meterDataList.Add(meterMap);
+                        }
+
+                        // 进水水质和出水水质
+                        string waterQuelitySql = @"SELECT TOP 1 [id]
+                                            ,[originWaterZD],[originWaterPH],[chendianchiOutWaterZD],[lvhouZD]
+                                            ,[lvhouYulv],[qingshuichiFlow],[gongshuishuitaOutFlow]
+                                            ,[outWaterZD],[outWaterPH],[outWaterYulv]
+                                            ,[qingshuichiFluidLevel],[CollectTime],[CreateDt]
+                                              FROM[dbo].[WaterFactoryDaliyRecord] where CollectTime >= '" + searchTime.ToString("yyyy-MM-dd HH:mm") + "' ";
+
+                        DataTable wqDt = waterFHelper.Fill(waterQuelitySql);
+                        if (wqDt != null && wqDt.Rows.Count > 0)
+                        {
+                            inPh = wqDt.Rows[0]["originWaterPH"];
+                            inChlorine = wqDt.Rows[0]["lvhouYulv"];
+                            inTurbidty = wqDt.Rows[0]["originWaterZD"];
+                            inTurbidtytwo = wqDt.Rows[0]["lvhouZD"];
+                            outPh = wqDt.Rows[0]["outWaterPH"];
+                            outChlorine = wqDt.Rows[0]["outWaterYulv"];
+                            outTurbidty = wqDt.Rows[0]["outWaterZD"];
+                            level = wqDt.Rows[0]["qingshuichiFluidLevel"];
+                        }
+                        Dictionary<string, object> meterMap1 = new Dictionary<string, object>(); // 进水
+                        meterMap1["InstantaneousFlow"] = null;
+                        meterMap1["NetCumulativeFlow"] = null;
+                        meterMap1["Pressure"] = null;
+                        meterMap1["PH"] = inPh;
+                        meterMap1["Chlorine"] = inChlorine;
+                        meterMap1["Turbidity"] = inTurbidty;
+                        meterMap1["TurbidityTwo"] = inTurbidtytwo;
+                        meterMap1["ReadTime"] = time;
+                        meterMap1["MeterCode"] = "wwkj008";
+                        meterDataList.Add(meterMap1);
+                        Dictionary<string, object> meterMap2 = new Dictionary<string, object>(); // 出水
+                        meterMap2["InstantaneousFlow"] = null;
+                        meterMap2["NetCumulativeFlow"] = null;
+                        meterMap2["PH"] = outPh;
+                        meterMap2["Chlorine"] = outChlorine;
+                        meterMap2["Turbidity"] = outTurbidty;
+                        meterMap2["TurbidityTwo"] = null;
+                        meterMap2["Pressure"] = null;
+                        meterMap2["ReadTime"] = time;
+                        meterMap2["MeterCode"] = "wwkj004";
+                        meterDataList.Add(meterMap2);
+
+                        baseMap["LiquidHeight"] = level;//baseRow["LiquidHeight"].ToString();
+
+                        #region 
+                        // 处理水厂查询的数据
+                        //for (int j = 0; j < dtMeter.Rows.Count; j++)
+                        //{
+                        //    string tableName = dtMeter.Rows[j]["TableName"].ToString();
+                        //    string meterId = dtMeter.Rows[j]["MeterId"].ToString();
+                        //    string meterCode = dtMeter.Rows[j]["MeterCode"].ToString();
+
+                        //    string colSql = "select * from 水厂数据字段 where MeterId = " + meterId;
+                        //    DataTable dtMeterCol = dbHelper.Fill(colSql);
+                        //    if (dtMeterCol == null || dtMeterCol.Rows.Count == 0)
+                        //    {
+                        //        continue;
+                        //    }
+                        //    string queryCol = "";
+                        //    string[] queryColArr = new string[dtMeterCol.Rows.Count];
+                        //    for (int k = 0; k < dtMeterCol.Rows.Count; k++)
+                        //    {
+                        //        string tableColumn = dtMeterCol.Rows[k]["TableColumn"].ToString();
+                        //        string sendKey = dtMeterCol.Rows[k]["SendKey"].ToString();
+                        //        if (k == dtMeterCol.Rows.Count - 1)
+                        //        {
+                        //            queryCol += tableColumn + " " + sendKey;
+                        //        }
+                        //        else
+                        //        {
+                        //            queryCol += tableColumn + " " + sendKey + ",";
+                        //        }
+                        //        queryColArr[k] = sendKey;
+                        //    }
+
+                        //    Dictionary<string, object> meterMap = new Dictionary<string, object>();
+                        //    string[] colArr = waterFactoryColumn.Split(',');
+                        //    foreach (string col in colArr)
+                        //    {
+                        //        meterMap[col] = null;
+                        //    }
+                        //    string meterDataSql = "select " + queryCol + " from " + tableName + " where 更新时间 = '" + id + "'";
+                        //    DataTable data = dbHelper.Fill(meterDataSql);
+                        //    if (data == null || data.Rows.Count == 0)
+                        //    {
+                        //        continue;
+                        //    }
+                        //    foreach (string col in queryColArr)
+                        //    {
+                        //        meterMap[col] = data.Rows[0][col];
+                        //    }
+                        //    meterMap["MeterCode"] = meterCode;
+                        //    meterDataList.Add(meterMap);
+                        //}
+                        #endregion
+
+                        factoryMap["BaseData"] = baseMap;
+                        factoryMap["MeterData"] = meterDataList;
+
+
+                        string message = JsonConvert.SerializeObject(factoryMap);
+
+
+                        foreach (KeyValuePair<string, IModel> item in channels)
+                        {
+                            string key = item.Key;
+                            IModel channel = item.Value;
+                            IBasicProperties property = properties[key];
+                            channel.BasicPublish("waterFactory.deviceHis", "", property, Encoding.UTF8.GetBytes(message)); //生产消息
+                        }
+
+                        // 删除数据
+                        String updatesql = "UPDATE [dbo].[uploadList] SET uploadTime = '" + time + "' WHERE [id] = " + dtFactory.Rows[i]["id"].ToString();
+                       
+                        waterFHelper.ExecuteNonQuery(updatesql);
+                    }
+                }
+
+                log.Info("水厂历史记录同步任务执行结束.................\r\n");
+            }
+            catch (Exception ex)
+            {
+                log.Error("水厂历史记录同步任务执行错误：" + ex.Message + "===========" + ex.StackTrace + "\r\n");
+            }
+        }
+
+        /// <summary>
+        /// 大表数据
+        /// </summary>
+        static IDbProvider dbHelper
+        {
+            get
+            {
+                var DbDefine = DbFactoryProvider.GetProvider(CurrentDbType.SqlServer, Constants.DbConncetion);
+                return DbDefine;
+            }
+        }
+
+        /// <summary>
+        /// 水厂数据库
+        /// </summary>
+        static IDbProvider waterFHelper
+        {
+            get
+            {
+                var DbDefine = DbFactoryProvider.GetProvider(CurrentDbType.SqlServer, Constants.changleWaterFactoryDb);
+                return DbDefine;
+            }
+        }
+    }
+}

TimedUpload/TimedUpload.csproj

@@ -94,7 +94,9 @@
     <Compile Include="Constants.cs" />
     <Compile Include="Program.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="QuartzJobs\ChangleBusinessDataJob.cs" />
     <Compile Include="QuartzJobs\ChangleSecondPumpDataJob.cs" />
+    <Compile Include="QuartzJobs\ChangleWaterFactoryDataJob.cs" />
     <Compile Include="QuartzJobs\DABusinessDataJob.cs" />
     <Compile Include="QuartzJobs\NoiseDataUploadJob.cs" />
     <Compile Include="QuartzJobs\TestJob.cs" />

TimedUpload/app.config

@@ -5,6 +5,7 @@
     <add key="WaterFactoryDbConncetion" value="Data Source=39.99.237.110;Initial Catalog=大安水厂;uid=sa;password=wwkj@2136807" />
     <add key="ChargeDB" value="Data Source=222.163.159.218,10433;Initial Catalog=ChargeManage;uid=sa;password=Daswwwkj@123" />
     <add key="zhihuishuiwuDB" value="server=39.99.237.110;user id=root;password=wwkj@2136807;database=smartwater_daan;charset=utf8" />
+    <add key="changleWaterFactoryDb" value=""/>
     <!-- 智慧水务系统RabbitMQ信息 start -->
     <add key="UploadUrl" value="127.0.0.1" />
     <add key="UploadPort" value="5678" />

TimedUpload/quartz_jobs.xml

@@ -169,19 +169,57 @@
     </trigger>-->
     <!--昌乐泵房数据定时上传数据-->
     <job>
-      <name>SecondaryPumpDataUploadJob</name>
-      <group>SecondaryPumpDataUploadData</group>
+      <name>ChangleSecondPumpDataJob</name>
+      <group>ChangleSecondPumpData</group>
       <description>昌乐泵房数据定时上传服务</description>
-      <job-type>TimedUpload.QuartzJobs.SecondaryPumpDataUploadJob,TimedUpload</job-type>
+      <job-type>TimedUpload.QuartzJobs.ChangleSecondPumpDataJob,TimedUpload</job-type>
       <durable>true</durable>
       <recover>false</recover>
     </job>
     <trigger>
       <cron>
-        <name>SecondaryPumpDataUploadJobbTrigger</name>
-        <group>SecondaryPumpDataUploadData</group>
-        <job-name>SecondaryPumpDataUploadJob</job-name>
-        <job-group>SecondaryPumpDataUploadData</job-group>
+        <name>ChangleSecondPumpDataJobTrigger</name>
+        <group>ChangleSecondPumpData</group>
+        <job-name>ChangleSecondPumpDataJob</job-name>
+        <job-group>ChangleSecondPumpData</job-group>
+        <start-time>2017-08-08T00:00:00+08:00</start-time>
+        <cron-expression>0 0/1 * * * ? *</cron-expression>
+      </cron>
+    </trigger>
+    <!--昌乐水厂数据定时上传数据-->
+    <job>
+      <name>ChangleWaterFactoryDataJob</name>
+      <group>ChangleWaterFactoryData</group>
+      <description>昌乐泵房数据定时上传服务</description>
+      <job-type>TimedUpload.QuartzJobs.ChangleWaterFactoryDataJob,TimedUpload</job-type>
+      <durable>true</durable>
+      <recover>false</recover>
+    </job>
+    <trigger>
+      <cron>
+        <name>ChangleWaterFactoryDataJobTrigger</name>
+        <group>ChangleWaterFactoryData</group>
+        <job-name>ChangleWaterFactoryDataJob</job-name>
+        <job-group>ChangleWaterFactoryData</job-group>
+        <start-time>2017-08-08T00:00:00+08:00</start-time>
+        <cron-expression>0 0 1 * * ? *</cron-expression>
+      </cron>
+    </trigger>
+    <!--昌乐营收数据定时上传数据-->
+    <job>
+      <name>ChangleBusinessDataJob</name>
+      <group>ChangleBusinessData</group>
+      <description>数据定时上传服务</description>
+      <job-type>TimedUpload.QuartzJobs.ChangleBusinessDataJob,TimedUpload</job-type>
+      <durable>true</durable>
+      <recover>false</recover>
+    </job>
+    <trigger>
+      <cron>
+        <name>ChangleBusinessDataJobTrigger</name>
+        <group>ChangleBusinessData</group>
+        <job-name>ChangleBusinessDataJob</job-name>
+        <job-group>ChangleBusinessData</job-group>
         <start-time>2017-08-08T00:00:00+08:00</start-time>
         <cron-expression>0 0 1 * * ? *</cron-expression>
       </cron>