zhihuishuiwuDataSyn/TimedUpload/QuartzJobs/WaterFactoryAreaDataJob.cs

using log4net;
using Quartz;
using RabbitMQ.Client;
using RDIFramework.Utilities;
using System;
using System.Collections.Generic;
using System.Data;
using System.IO;
using System.Text;

namespace TimedUpload.QuartzJobs
{
    [DisallowConcurrentExecution]
    public class WaterFactoryAreaDataJob :IJob
    {
        private readonly ILog log = LogManager.GetLogger(typeof(WaterFactoryAreaDataJob));

        public void Execute(IJobExecutionContext context)
        {
            // throw new NotImplementedException();
            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.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)
            {
                SendZoneDeviceHis(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 SendZoneDeviceHis(Dictionary<string, IModel> channels, Dictionary<string, IBasicProperties> properties)
        {
            try
            {
                log.Info("水厂大表设备历史数据同步任务开始执行.................\r\n");
                String sqlMeter = "SELECT * FROM bs_waterfactory_meter WHERE Type = 1 AND IsFlow = 1";//"SELECT ID,考核表编码 FROM [设备信息] where 是否启用 = '是' and 考核表编码 is not null order by ID";
                DataTable dtMeter = dbHelper.Fill(sqlMeter);

                //Dictionary<string, object> arguments = new Dictionary<string, object>();
                //arguments["x-max-length-bytes"] = 2147383648;
                //arguments["x-overflow"] = "reject-publish";

                
                for (int i = 0; i < dtMeter.Rows.Count; i++)
                {
                    DataRow drMeter = dtMeter.Rows[i];
                    //String meterId = drMeter["ID"].ToString();
                    String meterCode = drMeter["MeterCode"].ToString().TrimEnd() + "wf";
                    String lastTime = "";

                    //for (int k = lastYear; k <= nowYear; k++)
                    //{
                        //string tablename = "历史记录_" + ("000000" + meterId).Substring(meterId.Length, 6) + "_" + k;

                        // 判断历史记录表是否存在
                        //if (!CheckTableExist(tablename))
                        //{
                        //    continue;
                        //}

                        //String sqlMeterHis = "select 记录时间,采集时间,正累计流量,负累计流量,净累计流量,瞬时流量,电池电压,压力 from " + tablename;
                        //if (uploadHis.ContainsKey(meterCode))
                        //{
                        //    sqlMeterHis += " where 采集时间 > '" + uploadHis[meterCode] + "'";
                        //}
                        //sqlMeterHis += " order by 采集时间";
                        //DataTable dtMeterHis = dbHelper.Fill(sqlMeterHis);

                        StringBuilder message = new StringBuilder();
                        for (int j = 0; j < dtMeter.Rows.Count; j++)
                        {
                            message.Clear();
                            try
                            {
                                //DataRow drMeterHis = dtMeterHis.Rows[j];
                                String getDateTime = Convert.ToDateTime(drMeter["ReadTime"]).ToString("yyyy-MM-dd HH:mm:ss");
                                message.Append("{");
                                message.Append("\"meterAssessmentCode\": \"").Append(drMeter["MeterCode"].ToString()).Append("wf\",");
                                message.Append("\"manufacturerCode\": ").Append(Constants.ManufacturerCode).Append(",");
                                message.Append("\"getDateTime\": \"").Append(getDateTime).Append("\",");
                                if (Convert.DBNull != drMeter["NetCumulativeFlow"])
                                {
                                    message.Append("\"netCumulativeFlow\": ").Append(Convert.ToDecimal(drMeter["NetCumulativeFlow"])).Append(",");
                                }
                                if (Convert.DBNull != drMeter["NetCumulativeFlow"])
                                {
                                    message.Append("\"positiveCumulativeFlow\": ").Append(Convert.ToDecimal(drMeter["NetCumulativeFlow"])).Append(",");
                                }
                                
                                message.Append("\"negativeCumulativeFlow\": ").Append(0).Append(",");
                               
                                if (Convert.DBNull != drMeter["InstantaneousFlow"])
                                {
                                    message.Append("\"instantaneousFlow\": ").Append(Convert.ToDecimal(drMeter["instantaneousFlow"])).Append(",");
                                }
                                if (Convert.DBNull != drMeter["Pressure"])
                                {
                                    message.Append("\"pressure\": ").Append(Convert.ToDecimal(drMeter["Pressure"])).Append(",");
                                }
                                //if (Convert.DBNull != drMeter["电池电压"])
                                //{
                                //    message.Append("\"batteryVoltageValue\": ").Append(Convert.ToDecimal(drMeterHis["电池电压"])).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.deviceHis", "", property, Encoding.UTF8.GetBytes(message.ToString())); //生产消息
                                }
                                lastTime = getDateTime;
                            }
                            catch (Exception ex)
                            {
                                log.Info("大表设备历史记录同步任务数据推送失败：" + message.ToString() + "\r\n");
                                log.Error(ex.Message + "===========" + ex.StackTrace + "\r\n");
                            }
                        }
                    //}
                    //if (!"".Equals(lastTime))
                    //{
                    //    uploadHis[meterCode] = lastTime;
                    //}
                }

                //SavaUploadHis(uploadHis);
                log.Info("水厂大表设备历史记录同步任务执行结束.................\r\n");
            }
            catch (Exception ex)
            {
                log.Error("水厂大表设备历史记录同步任务执行错误：" + ex.Message + "===========" + ex.StackTrace + "\r\n");
            }
        }

        /// <summary>
        /// 判断历史记录表是否存在
        /// </summary>
        /// <param name="tablename"></param>
        /// <returns></returns>
        private bool CheckTableExist(string tablename)
        {
            DataTable table = dbHelper.Fill("select top 1 * from sysobjects where name='" + tablename + "' and xtype='u'");
            if (table == null || table.Rows.Count == 0)
            {
                return false;
            }
            return true;
        }

        /// <summary>
        /// 保存每块块表的上传最后一条历史记录
        /// </summary>
        /// <param name="uploadHis"></param>
        private void SavaUploadHis(Dictionary<String, String> uploadHis)
        {
            // 清除之前的内容
            FileStream stream = File.Open(@"TextFile1.txt", FileMode.OpenOrCreate, FileAccess.Write);
            stream.Seek(0, SeekOrigin.Begin);
            stream.SetLength(0);
            stream.Close();

            using (StreamWriter sw = new StreamWriter(@"TextFile1.txt"))
            {
                foreach (var item in uploadHis)
                {
                    sw.WriteLine(item.Key + "," + item.Value);
                }
            }
        }

        static IDbProvider dbHelper
        {
            get
            {
                var DbDefine = DbFactoryProvider.GetProvider(CurrentDbType.MySql, Constants.zhihuishuiwuDB);
                return DbDefine;
            }
        }
    }
}