NoSQL初探之人人都愛Redis：（3）使用Redis作為消息隊列服務場景應用案例
http://www.cnblogs.com/edisonchou/p/3825682.html
一、消息隊列場景簡介
“消息”是在兩臺計算機間傳送的數據單位。消息可以非常簡單，例如只包含文本字符串；也可以更復雜，可能包含嵌入對象。消息被發送到隊列中，“消息隊列”是在消息的傳輸過程中保存消息的容器。
在目前廣泛的Web應用中，都會出現一種場景：在某一個時刻，網站會迎來一個用戶請求的高峰期（比如：淘寶的雙十一購物狂歡節，12306的春運搶票節等），一般的設計中，用戶的請求都會被直接寫入數據庫或文件中，在高並發的情形下會對數據庫服務器或文件服務器造成巨大的壓力，同時呢，也使響應延遲加劇。這也說明了，為什麽我們當時那麽地抱怨和吐槽這些網站的響應速度了。當時2011年的京東圖書促銷，曾一直出現在購物車中點擊“購買”按鈕後一直是“Service is too busy
從京東當年的“Service is too busy”不難看出，高並發的用戶請求是網站成長過程中必不可少的過程，也是一個必須要解決的難題。在眾多的實踐當中，除了增加服務器數量配置服務器集群實現伸縮性架構設計之外，異步操作也被廣泛采用。而異步操作中最核心的就是使用消息隊列，通過消息隊列，將短時間高並發產生的事務消息存儲在消息隊列中，從而削平高峰期的並發事務，改善網站系統的性能。在京東之類的電子商務網站促銷活動中，合理地使用消息隊列，可以有效地抵禦促銷活動剛開始就開始大量湧入的訂單對系統造成的沖擊
記得我在實習期間，成都市XXXX局的一個價格信息采集發布系統項目中有一個采集任務發布的模塊，其中每個任務都是一個事務，這個事務中需要向數據庫中不斷地插入行，每個任務發布時都要往表中插入幾百行甚至幾千行的任務數據（比如價格采集日報，往往需要發布2-3年的任務數據，每一天都是一個任務，所以大約有2,3千行任務期號數據，還要發給很多個區縣的監測中心，因此數據庫寫操作量很大，更別說同時發布的並發操作），由於業務邏輯的處理比較復雜和往數據庫的寫操作量交大，所以在沒有采用消息隊列時點擊“發布”按鈕後往往需要等待1分鐘左右的時間才提示“發布成功”，用戶體驗極不友好。
這時，我們就可以使用消息隊列的思想來重構這個發布模塊，在用戶點擊“發布”按鈕後，系統只需要把往數據庫插入的這個事務信息插入到指定的任務發布消息隊列裏邊去（入隊操作，這裏一般有一臺獨立的消息隊列服務器來單獨存儲和處理），然後系統就可以立即對用戶的這個發布請求進行響應（比如給出一個發布成功的操作提示，這裏暫不考慮消息隊列服務操作失敗的情形，如果失敗了，可以考慮采用給用戶發送郵件、短信或站內消息，讓其重新進行發布操作）。
最後，消息隊列服務器中有一個進程單獨對消息隊列進行處理，首先判斷消息隊列中是否有待處理的消息，如果有，則將其取出（出隊操作，堅持“先進先出”的順序，保證事務的準確性）進行相應地處理（比如這裏是進行保存數據的操作，將數據插入到數據庫服務器中的指定數據庫裏邊，實質還是文件的IO操作）。就這樣，通過消息隊列將高並發用戶請求進行異步操作，然後一一對消息隊列進行出隊的同步操作，也避免了並發控制的難題。
說到這裏，大家可能會想到這尼瑪不就是生產者消費者模式麽？對的，麽麽嗒，消息隊列就是生產者消費者模式的典型場景。簡單地說，客戶端不同用戶發送的操作請求就是生產者，他們將要處理的事務存儲到消息隊列中，然後消息隊列服務器的某個進程不停地將要處理的單個事務從消息隊列中一個一個地取出來進行相應地處理，這就是消費者消費的過程。
下面我們將以異常日誌為案例，介紹在.Net中如何采用消息隊列的思想解決並發問題。當然，消息隊列只是解決並發問題的其中一種方式，在實際中往往需要結合多種不同的技術方式來共同解決，比如負載均衡、反向代理、集群等方案。這裏，雖然以異常日誌為案例，但是“麻雀雖小五臟俱全”，日誌寫入文件的高並發操作也同樣適用於數據庫的高並發，所以，研究這個案例是具有實際意義的。
二、使用預置類型實現異常日誌隊列
在日常的Web應用中，異常日誌的記錄是一個十分重要的要點。因為，人無完人，系統也一樣，難免會在什麽時候出一個測試階段未能完全測試到的異常。這時候，不能將異常信息直接顯示給客戶，那樣既不友好也不安全。所以，一般都采用將異常信息記錄到日誌文件中（比如某個txt文件，數據庫中某個表等），然後技術支持人員通過查看異常日誌，分析異常原因，改進BUG重新發布，保障系統正常運行。
在用戶的各種操作中，如果出現異常的時間一致，那麽記錄異常日誌的操作就會成為並發操作，而記錄異常日誌又屬於文件的IO操作（其實數據庫的讀寫歸根結底也是對文件即對磁盤進行的IO操作），因此很有可能帶來並發控制的一系列問題。在以往的編碼實踐中，我們可以通過給不同的IO請求進行加鎖（C#中的lock），等第一個請求完成寫入後釋放鎖，第二個請求再獲得鎖，進行IO操作，然後釋放掉，一直到第N個請求釋放後結束。這種方式，雖然解決了並發操作帶來的問題，但是通過加鎖延遲了用戶響應請求的時間（比如第一個正在IO寫入操作時，後面的均處於等待狀態），並且加鎖也會給服務器帶來一定的性能負擔，造成服務器性能的下降。
基於以上原因，我們采用消息隊列的思想將異常日誌的記錄操作改為隊列版，這裏我們先不采用Redis，直接使用.Net為我們提供的預置類型-Queue。接下來，就讓我們動手開刀，寫起來。
（1）新建一個ASP.NET MVC 4項目，選擇“基本”類型，視圖引擎選擇“Razor”。
（2）既然是異常日誌記錄，首先得有異常。這時，我們腦海中想到了那個經典的異常：DividedByZeroException。於是，在Controllers文件夾中新建一個Controller，取名為Home（這裏因為Global文件中的默認路由就指向了Home控制器中的Index這個Action），在HomeController中修改Index這個Action的代碼如下：
public ActionResult Index()
{
int a = 10;
int b = 0;
int c = a / b; //會拋一個DividedByZero的異常
return View();
}
（3）在ASP.NET MVC項目中，我們需要在Global.asax中的Application_Start這個事件中修改全局過濾器（主要是App_Start中的FilterConfig類的RegisterGlobalFilters這個方法），讓系統支持對異常的全局處理操作（我們這裏主要是對異常進行記錄到指定文件中）。PS：Application_Start是整個Web應用的起始事件，主要進行一些配置（如過濾器配置、日誌器配置、路由配置等等）的初始化操作，當然這些配置也只會進行一次。
public class FilterConfig
{
public static void RegisterGlobalFilters(GlobalFilterCollection filters)
{
// MyExceptionFilterAttribute繼承自HandleError，主要作用是將異常信息寫入日誌文件中
filters.Add(new MyExceptionFilterAttribute());
// 默認的異常記錄類
filters.Add(new HandleErrorAttribute());
}
}
通過改寫過濾器配置，我們向全局過濾器中註冊了一個異常處理的過濾器配置，那麽這個MyExceptionFilterAttribute類又是如何編寫的呢？
public class MyExceptionFilterAttribute : HandleErrorAttribute
{
//版本1：使用預置隊列類型存儲異常對象
public static Queue<Exception> ExceptionQueue = new Queue<Exception>();
public override void OnException(ExceptionContext filterContext)
{
//將異常信息入隊
ExceptionQueue.Enqueue(filterContext.Exception);
//跳轉到自定義錯誤頁
filterContext.HttpContext.Response.Redirect("~/Common/CommonError.html");
base.OnException(filterContext);
}
}
通過使該類繼承HandlerErrorAttribute並使其覆寫OnException這個事件，代表在異常發生時可以進行的操作。而我們在這兒主要通過一個異常隊列將獲取的異常寫入隊列，然後跳轉到自定義錯誤頁：~/Common/CommonError.html，這個錯誤頁很簡單，就是簡單的顯示“系統發生錯誤，5秒後自動跳轉到首頁”
View Code
（4）走到這裏，生產者消費者模式中生產者的任務已經完成了，接下來消費者就需要開始消費了。也就是說，消息隊列已經建好了，我們什麽時候從隊列中去任務，在哪裏執行？怎麽樣執行？通過上面的介紹，我們知道，在專門的消息隊列服務器中有一個進程在始終不停地監視消息隊列，如果有需要待辦的任務信息，則會立即從隊列中取出來執行相應的操作，直到隊列為空為止。於是，思路有了，我們馬上來實現以下。這個消息監視的操作也是一個全局操作，在系統啟動時就會一直運行，於是它也應該寫在Application_Start這個全局起始事件裏邊，於是按照標準的配置寫法，我們在Application_Start中添加了如下代碼：MessageQueueConfig.RegisterExceptionLogQueue()；
protected void Application_Start()
{
AreaRegistration.RegisterAllAreas();
WebApiConfig.Register(GlobalConfiguration.Configuration);
FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
RouteConfig.RegisterRoutes(RouteTable.Routes);
BundleConfig.RegisterBundles(BundleTable.Bundles);
//自定義事件註冊
MessageQueueConfig.RegisterExceptionLogQueue();
}
那麽，這個MessageQueueConfig.RegisterExceptionLogQueue()又是怎麽寫的呢？
public class MessageQueueConfig
{
public static void RegisterExceptionLogQueue()
{
string logFilePath = HttpContext.Current.Server.MapPath("/App_Data/");
//通過線程池開啟線程，不停地從隊列中獲取異常信息並將其寫入日誌文件
ThreadPool.QueueUserWorkItem(o =>
{
while (true)
{
try
{
if (MyExceptionFilterAttribute.ExceptionQueue.Count > 0)
{
Exception ex = MyExceptionFilterAttribute.ExceptionQueue.Dequeue(); //從隊列中出隊，獲取異常對象
if (ex != null)
{
//構建完整的日誌文件名
string logFileName = logFilePath + DateTime.Now.ToString("yyyy-MM-dd") + ".txt";
//獲得異常堆棧信息
string exceptionMsg = ex.ToString();
//將異常信息寫入日誌文件中
File.AppendAllText(logFileName, exceptionMsg, Encoding.Default);
}
}
else
{
Thread.Sleep(1000); //為避免CPU空轉，在隊列為空時休息1秒
}
}
catch (Exception ex)
{
MyExceptionFilterAttribute.ExceptionQueue.Enqueue(ex);
}
}
}, logFilePath);
}
}
現在，讓我們來看看這段代碼：
①首先定義Log文件存放的文件夾目錄，這裏我們一般放到App_Data裏邊，因為放到這裏邊外網是無法訪問到的，可以防止下載操作；
②其次通過線程池ThreadPool開啟一個線程，不停地監聽消息隊列裏邊的待辦事項個數，如果個數>0，則進行出隊（FIFO，先入隊的先出隊）操作。這裏主要是取出具體的異常實例對象，並將異常的具體堆棧信息追加寫入到指定命名格式的文件中。
PS：許多應用程序創建的線程都要在休眠狀態中消耗大量時間，以等待事件發生。其他線程可能進入休眠狀態，只被定期喚醒以輪詢更改或更新狀態信息。線程池通過為應用程序提供一個由系統管理的輔助線程池使您可以更為有效地使用線程。關於線程池的更多信息請訪問：http://msdn.microsoft.com/zh-cn/library/system.threading.threadpool(v=VS.90).aspx
③如果該線程檢測到消息隊列中無待辦事項，則使用Thread.Sleep使線程“休息”一會，避免了CPU空轉（從理論上來說，CPU資源是很珍貴的，應該盡量提高CPU的利用率）。
（5）最後，我們來看看效果如何？
①首先，高大上的VS捕捉到了異常-DividedByZeroException：
②按照我們的全局異常處理過濾器，會將此異常記入隊列中，並返回HTTP 302重定向跳轉到自定義錯誤頁面：
③最後，打開App_Data文件夾，查看日誌文件：
到這裏時，我們已經借助消息隊列的思想完成了一個自定義的異常日誌隊列服務。但也許有朋友會說，這個跟Redis有關系麽？異常日誌不都是用Log4Net麽？不要著急，後邊我們就會使用Redis+Log4Net來重構這個異常日誌隊列服務，不要走開，我們不得插播廣告哦，麽麽嗒！
三、使用Redis重構異常日誌隊列
（1）第一步，開啟Redis的服務，這裏我們使用命令開啟Redis服務（之前已經將Redis註冊到了Windows系統服務中了嘛，麽麽嗒）：net start redis-instance，當然，也可以通過在Windows服務列表中開啟。
（2）第二步，在剛剛的版本1的Demo中新建一個文件夾，命名為Lib，將ServiceStack.Redis的dll和Log4Net的dll都拷貝進去。然後，在引用中添加對Lib文件夾中所有dll的引用。
（3）第三步，重寫MyExceptionFilterAttribute這個全局異常信息過濾器。這裏使用到了Redis的客戶端連接池，每次連接時都是從池中取，不需要每次都創建，節省了時間和資源，提高了資源利用率。對於，多臺Redis服務器組成的集群而言，這裏需要指定多個形如 IP地址:端口號 的字符串數組。
public class MyExceptionFilterAttribute : HandleErrorAttribute
{
//版本2：使用Redis的客戶端管理器（對象池）
public static IRedisClientsManager redisClientManager = new PooledRedisClientManager(new string[]
{
//如果是Redis集群則配置多個{IP地址:端口號}即可
//例如: "10.0.0.1:6379","10.0.0.2:6379","10.0.0.3:6379"
"127.0.0.1:6379"
});
//從池中獲取Redis客戶端實例
public static IRedisClient redisClient = redisClientManager.GetClient();
public override void OnException(ExceptionContext filterContext)
{
//將異常信息入隊
redisClient.EnqueueItemOnList("ExceptionLog", filterContext.Exception.ToString());
//跳轉到自定義錯誤頁
filterContext.HttpContext.Response.Redirect("~/Common/CommonError.html");
base.OnException(filterContext);
}
}
（4）第四步，首先在Web.config中加入Log4Net的詳細配置。
View Code
PS:Log4Net是用來記錄日誌的一個常用組件（Log4J的移植版本），可以將程序運行過程中的信息輸出到一些地方（文件、數據庫、EventLog等）。由於Log4Net不是本篇博文介紹的重點，所以對Log4Net不熟悉的朋友，請在博客園首頁搜索：Log4Net，瀏覽其詳細的介紹。
其次，在App_Start文件夾中添加一個類，取名為LogConfig，定義一個靜態方法：RegisterLog4NetConfigure，具體代碼只有一行，實現了Log4Net配置的初始化操作。
public class LogConfig
{
public static void RegisterLog4NetConfigure()
{
//獲取Log4Net配置信息(配置信息定義在Web.config文件中)
log4net.Config.XmlConfigurator.Configure();
}
}
最後，在Global.asax中的Application_Start方法中添加一行代碼，註冊Log4Net的配置：
protected void Application_Start()
{
AreaRegistration.RegisterAllAreas();
WebApiConfig.Register(GlobalConfiguration.Configuration);
FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
RouteConfig.RegisterRoutes(RouteTable.Routes);
BundleConfig.RegisterBundles(BundleTable.Bundles);
//自定義事件註冊
MessageQueueConfig.RegisterExceptionLogQueue();
LogConfig.RegisterLog4NetConfigure();
}
（5）第五步，改寫MessageQueueConfig中的RegisterExceptionLogQueue方法。這裏就不再需要從預置類型Queue中取任務了，而是Redis中取出任務出隊進行相應處理。這裏，我們使用了Log4Net進行異常日誌的記錄工作。PS：註意在代碼頂部添加對log4net的引用：using log4net;
public static void RegisterExceptionLogQueue()
{
//通過線程池開啟線程，不停地從隊列中獲取異常信息並將其寫入日誌文件
ThreadPool.QueueUserWorkItem(o =>
{
while (true)
{
try
{
if (MyExceptionFilterAttribute.redisClient.GetListCount("ExceptionLog") > 0)
{
//從隊列中出隊，獲取異常對象
string errorMsg = MyExceptionFilterAttribute.redisClient.DequeueItemFromList("ExceptionLog");
if (!string.IsNullOrEmpty(errorMsg))
{
//使用Log4Net寫入異常日誌
ILog logger = LogManager.GetLogger("Log");
logger.Error(errorMsg);
}
}
else
{
Thread.Sleep(1000); //為避免CPU空轉，在隊列為空時休息1秒
}
}
catch (Exception ex)
{
MyExceptionFilterAttribute.redisClient.EnqueueItemOnList("ExceptionLog", ex.ToString());
}
}
});
}
（6）最後一步，調試驗證是否能正常寫入App_Data文件的日誌中，發現寫入的異常日誌如下，格式好看，信息詳細，圓滿完成了我們的目的。
四、小結
使用消息隊列將調用異步化，可以改善網站系統的性能：消息隊列具有很好的削峰作用，即通過異步處理，將短時間高並發產生的事務消息存儲在消息隊列中，從而削平高峰期的並發事務。在電商網站的促銷活動中，合理使用消息隊列，可以有效地抵禦促銷活動剛開始大量湧入的訂單對系統造成的沖擊。本文使用消息隊列的思想，借助Redis+Log4Net完成了一個超簡單的異常日誌隊列的應用案例，可以有效地解決在多線程操作中對日誌文件的並發操作帶來的一些問題。同樣地，借助消息隊列的思想，我們也可以完成對數據庫的高並發的消息隊列方案。所以，麻雀雖小五臟俱全，理解好了這個案例，相信對我們這些菜鳥碼農是有所裨益的。同樣，也請大牛們一笑而過，多多指教菜鳥們一步一步地提高，謝謝了！後邊，我們會探索一下Redis的集群、主從復制，以及在VMWare中建立幾臺虛擬機來構建主從結構，並使用Redis記錄網站中重要的Session會話對象，或者是電商項目中常見的商品類目信息等。但是，本人資質尚淺，並且都是一些初探性質的學習，如有錯誤和不當，還請各位園友多多指教！
參考文獻
（1）傳智播客.Net學院王承偉，數據優化技術之Redis公開課，http://bbs.itcast.cn/thread-26525-1-1.html
（2）Sanfilippo/賈隆譯，《幾點建議，讓Redis在你的系統中發揮更大作用》，http://database.51cto.com/art/201107/276333.htm
（3）NoSQLFan，《Redis作者談Redis應用場景》，http://blog.nosqlfan.com/html/2235.html
（4）善心如水，《C#中使用Log4Net記錄日誌》，http://www.cnblogs.com/wangsaiming/archive/2013/01/11/2856253.html
（5）逆心，《ServiceStack.Redis之IRedisClient》，http://www.cnblogs.com/kissdodog/p/3572084.html
（6）李智慧，《大型網站技術架構-核心原理與案例分析》，http://item.jd.com/11322972.html
附件下載
（1）版本1：使用預置類型的異常日誌隊列Demo，http://pan.baidu.com/s/1nt5G7Fj
（2）版本2：使用Redis+Log4Net的異常日誌隊列Demo，http://pan.baidu.com/s/1i3gMnnJ
NoSQL初探之人人都愛Redis：（3）使用Redis作為消息隊列服務場景應用案例