哎呀,我老大寫Bug啦——記一次MessageQueue的優化
MessageQueue,顧名思義訊息佇列,在系統開發中也是用的比較多的一箇中間件吧。我們這裡主要用它來做日誌管理和訂單管理的,記得老老大(恩,是的,就是老老大,因為他已經跳槽了)還在的時候,當時也是為了趕專案進度,他也參與開發了,那時候我才剛剛入職,他負責寫後端這塊,我來了就把他手上的任務接過來了,(接著接著……就辭職了)。
之後我們的開發仍然有條不紊的開發著,直到今年的一月份吧,才上線開始執行,然後就出現了常規狀態,上線之後就開始爆炸,
這個頁面打不開呀,那個內容沒東西呀,第三方登入問題呀,支付問題呀,臨時再改需求呀……(該來的都來了),加班、debug、測試、再debug……,然後經過幾天的修復,終於完成了跟自己電腦一樣穩定的執行,組員們都美滋滋的,今晚加個雞腿才行。
都說禍不單行,古人是不會騙我們的,Bug怎麼會修得完呢?天真,要是Bug能修得完還要我們來幹啥,好景不長,果然,過了一週之後,組員突然群裡叫喳喳,
what is it ?
來了,今天的主角登場了,我也要開始加班了。
這個是今天要說的東西,基礎概念什麼的不是今天要說的重點,重點是:
RabbitMQ記憶體暴漲!使得整個伺服器瀕臨癱瘓,遠端登入伺服器都差點擠不進去的狀態,別看截圖目前才1.3G,吃個午飯回來,就2.3G了,可怕不可怕?咋回事?
老闆喊你回來加班啦
先不管了,線上優先解決,手動先Reset回收資源以釋放空間,這個只是臨時的辦法,然後檢查一下rabbitMQ的配置有沒有問題,路徑在
C:\Users\Administrator\AppData\Roaming\RabbitMQ完全是預設的配置,完全ojbk啊,那到底咋回事?繼續檢查,想想不如從專案開始吧,然後檢視專案中的程式碼,都是從來自【MessageLib】的元件呼叫
好了,叫我老老大要這個元件的程式碼,他把git的地址就發給我,我把專案down下來,
這個封裝的元件內容不多,主要的檔案一目瞭然,其實就是用到這個兩個元件來進行的二次封裝來呼叫
主要的程式碼是在【MessageQueue.cs】檔案裡,展示一下當時的程式碼情況:
using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using MessageLib.ClassBean; using EasyNetQ; using System.Threading; namespace MessageLib { public static class MessageQueue { public static IBus bus = MQBusBuilder.CreateMessageBus(); //訊息佇列 private static Queue<Item> NoticQueue = new Queue<Item>(5000); //日誌佇列 private static Queue<Item> LogQueue = new Queue<Item>(5000); //佇列數目釋出數量 private static int max_count_to_pulish = 1000; /// <summary> /// 可供外部使用的訊息入列操作 /// </summary> public static void push(Item item) { if (item.type == ItemType.notic) { NoticQueue.Enqueue(item); } if (item.type == ItemType.log) { LogQueue.Enqueue(item); } } /// <summary> /// 監聽後需要呼叫的釋出介面 /// </summary> private static void Pulish(object source, System.Timers.ElapsedEventArgs e) { if (NoticQueue.Count > 0 || LogQueue.Count > 0) { if (bus == null || !bus.IsConnected) { bus = MQBusBuilder.CreateMessageBus(); } if (bus.IsConnected) { Send(ItemType.notic); Send(ItemType.log); } } } /// <summary> /// 程式自執行並開始監聽 /// </summary> public static void Run() { System.Timers.Timer timer = new System.Timers.Timer(); timer.Interval = 1000; timer.Elapsed += new System.Timers.ElapsedEventHandler(Pulish);//到達時間的時候執行事件; timer.AutoReset = true;//設定是執行一次(false)還是一直執行(true); timer.Enabled = true;//是否執行System.Timers.Timer.Elapsed事件; } /// <summary> /// 啟動執行緒非同步呼叫 /// </summary> /// <param name="channelType"></param> private static void Send(string channelType) { Thread thread = new Thread(new ParameterizedThreadStart(PublishAction)); thread.IsBackground = true; thread.Start(channelType); } /// <summary> /// 呼叫釋出日誌及提醒兩個介面 /// </summary> /// <param name="channel"></param> private static void PublishAction(object channel) { PublisLog(); PublisNotic(); } /// <summary> /// 日誌訊息傳送至RabbitMQ指定exchange、Queue /// </summary> private static void PublisLog() { string channelName = ItemType.log; try { var routingKey = channelName; var mqqueue = bus.Advanced.QueueDeclare(string.Format("Queue.{0}", channelName)); var exchange = bus.Advanced.ExchangeDeclare(string.Format("Exchange.{0}",channelName), "direct"); var binding = bus.Advanced.Bind(exchange, mqqueue, routingKey); while (LogQueue.Count > 0) { Item item = LogQueue.Dequeue(); if (item != null) { var properties = new MessageProperties(); var Message = new Message<string>(Newtonsoft.Json.JsonConvert.SerializeObject(item)); Message.Properties.AppId = item.appid; bus.Advanced.Publish(exchange, routingKey, false, Message); } } } catch (Exception ex) { throw ex; } } /// <summary> /// 提醒訊息傳送至RabbitMQ指定exchange、Queue /// </summary> private static void PublisNotic() { string channelName = ItemType.notic; var routingKey = channelName; var mqqueue = bus.Advanced.QueueDeclare(string.Format("Queue.{0}", channelName)); var exchange = bus.Advanced.ExchangeDeclare(string.Format("Exchange.{0}", channelName), "direct"); var binding = bus.Advanced.Bind(exchange, mqqueue, routingKey); while(NoticQueue.Count > 0) { Item item = NoticQueue.Dequeue(); if (item != null) { var properties = new MessageProperties(); var Message = new Message<string>(Newtonsoft.Json.JsonConvert.SerializeObject(item)); Message.Properties.AppId = item.appid; bus.Advanced.Publish(exchange, routingKey, false, Message); } } } } }View Code
然後我就發現了這一段程式碼!
/// <summary> /// 程式自執行並開始監聽 /// </summary> public static void Run() { System.Timers.Timer timer = new System.Timers.Timer(); timer.Interval = 1000; timer.Elapsed += new System.Timers.ElapsedEventHandler(Pulish);//到達時間的時候執行事件; timer.AutoReset = true;//設定是執行一次(false)還是一直執行(true); timer.Enabled = true;//是否執行System.Timers.Timer.Elapsed事件; }
/// <summary> /// 啟動執行緒非同步呼叫 /// </summary> /// <param name="channelType"></param> private static void Send(string channelType) { Thread thread = new Thread(new ParameterizedThreadStart(PublishAction)); thread.IsBackground = true; thread.Start(channelType); }
老老大寫Bug了,當Run()起來之後,佇列中【NoticQueue】有內容,就開始推送訊息,傳送訊息Send(),每來一次推送new一個執行緒並設定為後臺執行緒,然後傳送訊息。好了,明白了,這裡的執行緒很混亂,因為執行緒操作不當,new了N多個頻道,並且沒有主動回收,這也難怪記憶體暴漲呢。並且要是Run()呼叫多次,後果更加不堪設想。
加班改起來
開始動手吧,業務主要推送有普通訊息、錯誤訊息和通知訊息,那麼將佇列與執行緒組裝一起,新增一個類QueueTask.cs:
public class QueueTask { private Queue<Item> NoticQueue = new Queue<Item>(5000); //佇列數目釋出數量 private int max_count_to_pulish = 1000; public bool isRunning = false; private string itemType = ItemType.info; private string MessageRouter = ItemType.info; public QueueTask(string itemType,string MessageRouter) { this.itemType = itemType; this.MessageRouter = MessageRouter; } /// <summary> /// 可供外部使用的訊息入列操作 /// </summary> public void Push(Item item, IBus IBus) { NoticQueue.Enqueue(item); if (!isRunning) Run(IBus); } public void Run(IBus IBus) { if (!isRunning) { Timer timer = new Timer(PulishMsg, IBus, 1000, 1000);
timer.AutoReset = timer.Enabled = true; isRunning = true; } } private void PulishMsg(object state) { IBus IBus = state as IBus; if (NoticQueue.Count > 0) { PublisMsg(itemType, IBus); } } private void PublisMsg(object channel, IBus BusInstance) { try { string channelName = channel as string; if (NoticQueue.Count > 0) { var routingKey = MessageRouter; var mqqueue = BusInstance.Advanced.QueueDeclare(string.Format("Queue.{0}", channelName)); var exchange = BusInstance.Advanced.ExchangeDeclare(string.Format("Exchange.{0}", channelName), "direct"); var binding = BusInstance.Advanced.Bind(exchange, mqqueue, routingKey); while (NoticQueue.Count > 0) { Item item = NoticQueue.Dequeue(); if (item != null) { var properties = new MessageProperties(); var Message = new EasyNetQ.Message<string>(Newtonsoft.Json.JsonConvert.SerializeObject(item)); Message.Properties.AppId = item.appid; BusInstance.Advanced.Publish(exchange, routingKey, false, Message); } } } } catch (Exception ex) { Console.WriteLine("PublisMsg error:" + ex.Message); } } public void Read<T>(IBus BusInstance,Action<Item> dealAction) where T : Item { try { string channelName = itemType; var routingKey = MessageRouter; var mqqueue = BusInstance.Advanced.QueueDeclare(string.Format("Queue.{0}", channelName)); var exchange = BusInstance.Advanced.ExchangeDeclare(string.Format("Exchange.{0}", channelName), "direct"); var binding = BusInstance.Advanced.Bind(exchange, mqqueue, routingKey); var Consume = BusInstance.Advanced.Consume(mqqueue, registration => { registration.Add<string>((message, info) => { Item data = Newtonsoft.Json.JsonConvert.DeserializeObject<T>(message.Body); dealAction(data); }); }); } catch (Exception ex) { Console.WriteLine("Read error:" + ex.Message); } } }
然後,在MessageQueue.cs修改為單例模式:
public static class MessageQueue { /*Install-Package EasyNetQ-dotnet-core -Version 2.0.2-radicalgeek-netc0001 -Pre*/ private static IBus bus = null; public static bool isRunning = false; //訊息佇列 private static QueueTask NoticQueue = null; //日誌佇列 private static QueueTask LogQueue = null; //自定義 private static QueueTask InfoQueue = null; #region 同步鎖 private static readonly object obj = new object(); #endregion public static void Init(string Connection, string routeKey) { if (NoticQueue == null) NoticQueue = new QueueTask(ItemType.notic, ItemType.notic); if (LogQueue == null) LogQueue = new QueueTask(ItemType.error, ItemType.error); if (InfoQueue == null) InfoQueue = new QueueTask(ItemType.info, routeKey); if (string.IsNullOrEmpty(MQBusBuilder.Connnection)) MQBusBuilder.Connnection = Connection; } public static IBus BusInstance { get { if (bus == null) { lock (obj) { if (bus == null|| !bus.IsConnected) { bus = MQBusBuilder.CreateMessageBus(); } } } return bus; } } /// <summary> /// 可供外部使用的訊息入列操作 /// </summary> public static void PushAndRun(Item item) { if (string.IsNullOrWhiteSpace(MQBusBuilder.Connnection) || BusInstance == null) return; if (item.type == ItemType.notic) { NoticQueue.Push(item, BusInstance); } if (item.type == ItemType.error) { LogQueue.Push(item, BusInstance); } if (item.type == ItemType.info) { InfoQueue.Push(item, BusInstance); } } public static void Read(string itemType, Action<Item> dealAction) { if (itemType == ItemType.notic) { NoticQueue.Read<NoticItem>(BusInstance, dealAction); } if (itemType == ItemType.error) { LogQueue.Read<ErrorItem>(BusInstance, dealAction); } if (itemType == ItemType.info) { InfoQueue.Read<Message>(BusInstance, dealAction); } } }View Code
每次推送訊息的時候,每個QueueTask就自己維護自己的執行緒和隊列了,當呼叫推送之後,就開始運作起來。恩,應該沒問題了。然後就釋出nuget,再更新專案,然後釋出。觀察一段時間,恩,完美。
事件二
事情過後,B端開始搞起來了,然後涉及到訂單系統,跟老大(不是老老大,老老大那時候已經跑了)商量之後確定使用訊息佇列來做訂單的事件的拓展,然後就直接美滋滋的呼叫好之前寫的了,沒想到啊,這次是執行緒暴漲!因為訂單是從B端推送過來的,B端肯定沒事,訂單後臺訂閱訊息之後,讀取過程中出現的執行緒增多,然後看看之前寫的Read()方法,感覺沒啥問題啊,每執行完一次,就多了一個執行緒,這個神奇了啊,那麼原始碼擼起來。
翻來覆去,看到這個Consume方法,繼承的是IDisposable介面,得勒,知道咋回事了。
Consume.Dispose(); 用完請記得主動釋放啊。
這回真的可以浪了。
總結
遇到問題,冷靜下來,耐得了寂寞才行。線上的問題優先解決,然後再慢慢Debug,解決不了,看原始碼,再解決不了,降級處理,歡迎共同探討。同時也感謝一下技術群裡的兄弟給的一些建議,並幫忙查詢資料,還好EasyNetQ是開源了,不然也打算說先不用了,畢竟一開始沒什麼使用者量,所以沒必要整那麼麻煩,加班加點的弄這個問題。不過最終都完美的解決了,心裡還是挺美滋滋的,程式猿隨之而來的成就感。
別看我們在工位上默不作聲,我們可能在拯救世界呢!老闆,該加工資啦!