c# - MSMQ System.Messaging 资源使用率高

Question

我已经制作了 C# 控制台应用程序，它使用每 10 秒连接到 MSMQ 的计时器将数据插入 Oracle 数据库。但问题是它登录和注销域并创建高 CPU 也创建了非常多的安全审计日志，这浪费了我的资源。

我的控制台应用程序使用任务计划运行。代码如下

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Messaging;
using System.Xml;
using System.IO;
using System.Timers;
using Oracle.DataAccess.Client; 
using System.Data;


namespace MSMQ_News
{
    class Program
    {
    private static System.Timers.Timer aTimer;

    static void Main(string[] args)
    {
        try
        {

            // Create a timer with a ten second interval.
            aTimer = new System.Timers.Timer(60000);//10000
            // Hook up the Elapsed event for the timer.
            aTimer.Elapsed += new ElapsedEventHandler(OnTimedEvent);
            // Set the Interval to 2 seconds (2000 milliseconds).
            //aTimer.Interval = 10000;
            aTimer.Enabled = true;
            aTimer.Start();

            Console.WriteLine("Press the Enter key to exit the program.");
            Console.ReadLine();
        }
        catch (Exception ex)
        {
            Log(" From Main -- " + ex.Message);
        }
    }

    private static void OnTimedEvent(object source, ElapsedEventArgs e)
    {
        // Just in case someone wants to inherit your class and lock it as well ...
        object _padlock = new object();
        try
        {
            aTimer.Stop();
            lock (_padlock)
            {
                Console.WriteLine("The Elapsed event was raised at {0}", e.SignalTime);
                ProcessQueueMsgs();
            }
        }
        catch (Exception ex)
        {
            Log(" From OnTimedEvent -- " + ex.Message);
        }
        finally
        {
            aTimer.Start();
        }
    }

    private static void ProcessQueueMsgs()
    {
        try
        {
            while ((DateTime.Now.Hour >= 06)
                && (DateTime.Now.Hour <= 16))
            {
                DateTime dt = DateTime.Now;
                ReceiveNewsDetail(dt);
                ReceiveNewsHeader(dt);
            }
            CloseApp();
        }
        catch (Exception ex)
        {
            Log(" From ProcessQueueMsgs -- " + ex.Message);
        }
    }

    static bool QueueExist(string QueueName)
    {
        try
        {
            if (MessageQueue.Exists(QueueName))
                return true;
            else
                return false;
        }
        catch (Exception ex)
        {
            Log(" From QueueExist -- " + ex.Message);
            return false;
        }
    }

    private static void ReceiveNewsHeader(DateTime dt)
    {
        try
        {
            MessageQueue mqNewsHeader = null;
            string value = "", _tmp = "";
            _tmp = "<newsHeader></newsHeader> ";
            /*if (QueueExist(@".\q_ws_ampnewsheaderrep"))*/
            mqNewsHeader = new MessageQueue(@".\q_ws_ampnewsheaderrep");

            int MsgCount = GetMessageCount(mqNewsHeader, @".\q_ws_ampnewsheaderrep");
            for (int i = 0; i < MsgCount; i++)
            {
                Message Msg = mqNewsHeader.Receive();
                Msg.Formatter = new ActiveXMessageFormatter();

                //need to do this to avoid ??? for arabic characters
                using (StreamReader strdr = new StreamReader(Msg.BodyStream, System.Text.Encoding.Default))
                {
                    value = strdr.ReadToEnd();
                }

                value = value.Replace("\0", String.Empty);
                if (value != _tmp)
                {
                    LoadNewsHeader(value, dt);
                }
            }
        }
        catch (Exception ex)
        {
            Log("From ReceiveNewsHeader -- " + ex.Message);
        }
    }

    private static void ReceiveNewsDetail(DateTime dt)
    {
        try
        {
            MessageQueue mqNewsDetails = null;
            string value = "", _tmp = "";
            _tmp = "<news></news> ";
            /*if (QueueExist(@".\q_ws_ampnewsrep"))*/
                mqNewsDetails = new MessageQueue(@".\q_ws_ampnewsrep");

            int MsgCount = GetMessageCount(mqNewsDetails, @".\q_ws_ampnewsrep");
            for (int i = 0; i < MsgCount; i++)
            {
                Message Msg = mqNewsDetails.Receive();
                Msg.Formatter = new ActiveXMessageFormatter();

                //need to do this to avoid ??? for arabic characters
                using (StreamReader strdr = new StreamReader(Msg.BodyStream, System.Text.Encoding.Default))
                {
                    value = strdr.ReadToEnd();
                }

                value = value.Replace("\0", String.Empty);
                if (value != _tmp)
                {
                    LoadNewsDetail(value, dt);
                }
            }

        }
        catch (Exception ex)
        {
            Log("From ReceiveNewsDetail -- " + ex.Message);
        }
    }

    private static void LoadNewsHeader(string text , DateTime dt)
    {
        try
        {

            //text = ReplaceSpecialCharacters(text);
            //text = Clean(text);
            //XmlDocument _xmlDoc = new XmlDocument();
            //_xmlDoc.LoadXml(text);
            //string fileName = "NewsHeader.xml";
            text = text.Replace("<arabicFields>", "<arabicFields>\n\t\t");

            //createXMLFile(fileName, text);
            XmlDocument _xmlDoc = LoadXMLDoc(text);

            string SQL = "";


            XmlNodeList newsHeaderList = _xmlDoc.SelectNodes("newsHeader/newsHeaderRep");
            if (newsHeaderList.Count > 0)
            {
                OracleParameter pTRUNCATE = new OracleParameter("P_TABLE_NAME", OracleDbType.Varchar2);
                pTRUNCATE.Value = "COMPANIES_NEWS";
                DatabaseOperation(CommandType.StoredProcedure, "TRUNCATE_TABLE", pTRUNCATE);
            }
            foreach (XmlNode news in newsHeaderList)
            {

                XmlNodeList newsIdList = news.SelectNodes("newsId");
                SQL = "Insert into COMPANIES_NEWS(NewsID, NewsID_SEQNO, NEWSSTATUS, LANGUAGE_CD, SEC_CD, RELEASEDATE, RELEASETIME, TITLE, STG_TIME) Values(";
                foreach (XmlNode newsId in newsIdList)
                {
                    SQL += "'" + newsId["id"].InnerText + "',";
                    SQL += "" + newsId["seqNo"].InnerText + ",";
                }

                SQL += "'" + news["newsStatus"].InnerText + "',";

                XmlNodeList newsItemList = news.SelectNodes("newsItem");
                foreach (XmlNode newsItem in newsItemList)
                {
                    SQL += "'" + newsItem["languageId"].InnerText + "',";
                    if (newsItem["reSecCode"] != null)
                        SQL += "'" + newsItem["reSecCode"].InnerText + "',";
                    else
                        SQL += "' ',";
                    XmlNodeList releaseTimeList = newsItem.SelectNodes("releaseTime");
                    foreach (XmlNode releaseTime in releaseTimeList)
                    {
                        SQL += "TO_DATE('" + releaseTime["date"].InnerText + "','YYYYMMDD'),";
                        SQL += "" + releaseTime["time"].InnerText + ",";
                    }
                }

                XmlNodeList arabicFieldsList = news.SelectNodes("arabicFields");
                foreach (XmlNode arabicFields in arabicFieldsList)
                {
                    SQL += "'" + RevertSpecialCharacters(arabicFields["title_AR"].InnerText) + "',";
                }
                SQL += "TO_DATE('" + dt.ToString() + "','MM/DD/YYYY HH12:MI:SS PM'))";
                DatabaseOperation(CommandType.Text, SQL, null);
                Console.WriteLine("Header : " + DateTime.Now.ToString());
            }

            if (SQL != "") //RecordCount("Select Count(*) from COMPANIES_NEWS_DETAILS") > 0
            {
                OracleParameter pREFRESH = new OracleParameter("P_TABLE_NAMEs", OracleDbType.Varchar2);
                pREFRESH.Value = "COMPANIES_NEWS";

                DatabaseOperation(CommandType.StoredProcedure, "REFRESH_VW_ALL", pREFRESH);
            }
        }
        catch (Exception ex)
        {
            Log("From LoadNewsHeader -- " + ex.Message);
        }
    }

    private static void LoadNewsDetail(string text, DateTime dt)
    {
        try
        {

            //string fileName = "NewsDetail.xml";
            text = text.Replace("<arabicFields>", "<arabicFields>\n\t\t");

            //text = createXMLFile(fileName);
            //text = text.Replace("<arabicFields>", "<arabicFields>\n\t\t");
            XmlDocument _xmlDoc = LoadXMLDoc(text);


            string SQL = "";

            XmlNodeList newsList = _xmlDoc.SelectNodes("news/newsRep");
            if (newsList.Count > 0)
            {
                OracleParameter pTRUNCATE = new OracleParameter("P_TABLE_NAME", OracleDbType.Varchar2);
                pTRUNCATE.Value = "COMPANIES_NEWS_DETAILS";
                DatabaseOperation(CommandType.StoredProcedure, "TRUNCATE_TABLE", pTRUNCATE);
            }
            foreach (XmlNode news in newsList)
            {

                XmlNodeList newsIdList = news.SelectNodes("newsId");
                SQL = "Insert into Companies_news_details(NewsID_ID, NewsID_SEQNO, NewsText_1,NewsText_2,STG_TIME) Values(";
                foreach (XmlNode newsId in newsIdList)
                {
                    SQL += "" + newsId["id"].InnerText + ",";
                    SQL += "" + newsId["seqNo"].InnerText + ",";
                }
                XmlNodeList arabicFieldsList = news.SelectNodes("arabicFields");
                foreach (XmlNode arabicFields in arabicFieldsList)
                {
                   // Log(" Before Arabic Text Data -- :" + arabicFields["newsText_AR"].InnerText);
                    if (arabicFields["newsText_AR"].InnerText.Length > 4000)
                    {
                        SQL += "'" + RevertSpecialCharacters(arabicFields["newsText_AR"].InnerText.Substring(0, 3999)).Replace("\n",Environment.NewLine) + "',";
                        SQL += "'" + RevertSpecialCharacters(arabicFields["newsText_AR"].InnerText.Substring(3999, arabicFields["newsText_AR"].InnerText.Length)).Replace("\n", Environment.NewLine) + "',";
                        SQL += "TO_DATE('" + dt.ToString() + "','MM/DD/YYYY HH12:MI:SS PM')";
                    }
                    else
                    {
                        SQL += "'" + RevertSpecialCharacters(arabicFields["newsText_AR"].InnerText).Replace("\n", Environment.NewLine) + "','',";
                        SQL += "TO_DATE('" + dt.ToString() + "','MM/DD/YYYY HH12:MI:SS PM')";

                    }

                    SQL += ")";
                    DatabaseOperation(CommandType.Text, SQL, null);
                    Console.WriteLine("Detail : " + DateTime.Now.ToString());
                }
            }

            if (SQL != "") //RecordCount("Select Count(*) from COMPANIES_NEWS_DETAILS") > 0
            {
                OracleParameter pREFRESH = new OracleParameter("P_TABLE_NAMEs", OracleDbType.Varchar2);
                pREFRESH.Value = "COMPANIES_NEWS_DETAILS";

                DatabaseOperation(CommandType.StoredProcedure, "REFRESH_VW_ALL", pREFRESH);
            }
        }
        catch (Exception ex)
        {
            Log("From LoadNewsDetail -- " + ex.Message);
        }
    }

    private static void CloseApp()
    {
        System.Environment.Exit(0);    
    }

    protected static int GetMessageCount(MessageQueue q, string queueName)
    {
       var _messageQueue = new MessageQueue(queueName, QueueAccessMode.Peek);
       _messageQueue.Refresh();  //done to get the correct count as sometimes it sends 0
        var x = _messageQueue.GetMessageEnumerator2();
        int iCount = 0;
        while (x.MoveNext())
        {
           iCount++;
        }
        return iCount;
    }

    private static void DatabaseOperation(CommandType cmdType, string SQL, OracleParameter param)
    {
        string oracleConnectionString = System.Configuration.ConfigurationSettings.AppSettings["OracleConnectionString"];

        using (OracleConnection con = new OracleConnection())
        {
            con.ConnectionString = oracleConnectionString;
            con.Open();

            OracleCommand command = con.CreateCommand();
            command.CommandType = cmdType;
            command.CommandText = SQL;
            if (param != null)
                command.Parameters.Add(param);
            command.ExecuteNonQuery();

            command.Dispose();
            con.Close();
        }
    }

    private static String RevertSpecialCharacters(string pValue)
    {
        string _retVal = String.Empty;
        _retVal = pValue.Replace("'", "''");

        return _retVal;
    }

    public static void Log(string Message)
    {
        // Create a writer and open the file:
        StreamWriter log;
        //C:\Software\MSMQ_New_News_Fix
        if (!File.Exists(@"C:\MSMQ_New_News_Fix\log.txt"))
        {
            log = new StreamWriter(@"C:\MSMQ_New_News_Fix\log.txt");
        }
        else
        {
            log = File.AppendText(@"C:\MSMQ_New_News_Fix\log.txt");
        }

        // Write to the file:
        log.WriteLine(DateTime.Now.ToString() + " : " + Message);
        // Close the stream:
        log.Close();
    }

    public static XmlDocument LoadXMLDoc(string xmlText)
    {
        XmlDocument doc = new XmlDocument();
        try
        {
            string xmlToLoad = ParseXMLFile(xmlText);
            doc.LoadXml(xmlToLoad);
        }
        catch (Exception ex)
        {
            Log("From LoadXMLDoc -- " + ex.Message);
        }
        return doc;
    }

    private static string ParseXMLFile(string xmlText)
    {
        StringBuilder formatedXML = new StringBuilder();
        try
        {
            StringReader xmlReader = new StringReader(xmlText);
            while (xmlReader.Peek() >= 0)
                formatedXML.Append(ReplaceSpecialChars(xmlReader.ReadLine()) + "\n");
        }
        catch (Exception ex)
        {
            Log("From ParseXMLFile -- " + ex.Message);
        }
        return formatedXML.ToString();
    }

    private static string ReplaceSpecialChars(string xmlData)
    {
        try
        {
            //if (xmlData.Contains("objectRef")) return "<objectRef></objectRef>";
            int grtrPosAt = xmlData.IndexOf(">");
            int closePosAt = xmlData.IndexOf("</");
            int lenthToReplace = 0;
            if (grtrPosAt > closePosAt) return xmlData;

            lenthToReplace = (closePosAt <= 0 && grtrPosAt <= 0) ? xmlData.Length : (closePosAt - grtrPosAt) - 1;
            //get the string between xml element. e.g. <ContactName>Hanna Moos</ContactName>, 
            //you will get 'Hanna Moos'
            string data = xmlData.Substring(grtrPosAt + 1, lenthToReplace);
            string formattedData = data.Replace("&", "&amp;").Replace("<", "&lt;")
                                       .Replace(">", "&gt;").Replace("'", "&apos;");
            if (lenthToReplace > 0) xmlData = xmlData.Replace(data, formattedData);
            return xmlData;
        }
        catch (Exception ex)
        {
            Log("From ReplaceSpecialChars -- " + ex.Message);
            return "";
        }
    }

}
}

我该如何解决上述问题

score 0 · Accepted Answer

好吧，从逻辑上讲，您非常正确，我应该使 Windows 服务而不是计时器服务和任务计划。

但是我的问题是为什么经常登录/注销，这浪费了我的域服务器的资源。经过深入调查，我发现调用QueueExits是资源关键。我发现的另一件事是，当您连接 MSMQ 队列时，您正在登录共享资源，该资源将登录到域。由于我的代码每 10-20 秒运行一次，因此浪费了我的域服务器资源。

为了解决，我通过以下方式全局创建我的 MessageQueue 对象

private static MessageQueue mqNewsHeader = new MessageQueue(@".\q_ws_ampnewsheaderrep");
private static MessageQueue mqNewsDetails = new MessageQueue(@".\q_ws_ampnewsrep");

所以它将在应用程序的生命周期中创建一次，我们将只登录和注销一次。然后我会将这个对象作为参数传递给函数。我还看到我的 MessageQueue 计数函数也是资源关键，所以我将其更改为以下

protected static int GetMessageCount(MessageQueue q)
    {
       //var _messageQueue = new MessageQueue(queueName, QueueAccessMode.Peek);
       //_messageQueue.Refresh();  //done to get the correct count as sometimes it sends 0
       // var x = _messageQueue.GetMessageEnumerator2();
      int iCount = q.GetAllMessages().Count(); 
      // while (x.MoveNext())
       // {
       //    iCount++;
       // }

        return iCount;
    }

希望这能清楚我的答案，也能帮助其他人。

score 0 · Accepted Answer

为什么不在 Windows 服务中托管您的队列读取器进程。这将每 10 秒不断轮询队列。

然后使用 windows 调度程序在相关时间启动/停止服务以创建您的服务窗口。

这意味着您不需要在计划任务中做任何复杂的事情，也不会一直在加载和卸载。

c# - MSMQ System.Messaging 资源使用率高

2 回答 2

Related

Reference