在本章中我們將覆蓋:
- 使用Spring來開發web監控程序
- 使用Spring來開發異步web搜索
- 使用STOMP來開發web監控程序
介紹 RabbitMQ可以像客戶端一樣使用在服務端。當前，RabbitMQ覆蓋了大部分使用的語言和技術來構建web程序，如PHP,Node.js, Python, Ruby, 以及其它.你可以在http://www.rabbitmq.com/devtools.html找到全部的列表.
在本章中，我們將展示三種使用場景，如下所示，在這里RabbitMQ將被用于:
- 用于WebSockets通知的源(使用Spring來開發web監控程序食譜)
- 用于管理搜索結果的后端服務(使用Spring來開發異步web搜索食譜)
- 在web頁面上直接用來處理消息(使用STOMP食譜來開發web監控程序食譜)
在本章節中，我們選擇Spring Framework 來構建大多數例子。我們將展示一些有趣的特性，集成和RabbitMQ監控.
使用Spring來開發web監控程序
在本食譜中，我們將展示如何一個 web程序來監控服務器CPU負載和內存使用情況.本例中的服務器狀態是由a .NET客戶端來發送，但你可以使用其它語言或操作系統來擴展這些例子。它可以使用標準的JSON協議來發送消息。 例如，一個簡單的JSON編碼消息看起來像下面的文本:
{"UPDATETIME":"23/06/2013 22:55:32","SERVERID":"1","CPU":10,"MEM":40}
現在我們將介紹Spring Insight.使用Insight,你可以監控web 程序的性能和正確行為.
查看http://gopivotal.com/products/pivotal-tc-server來了解更多信息。
在客戶端，我們可以使用JQuery,Twitter的Bootstrap,以及Google Chart.
在這個例子的末尾，你可以在監控控制臺看到每個服務器的工作情況以及動態更新信息，如下面的截圖所示：
你可在Chapter05/Recipe01/web and the client at Chapter05/Recipe01/csharp中找到web程序的源碼.
準備
要讓例子能工作，你需要1.7+.
為了構建web程序,我們假設你使用Spring Tool Suite (也稱為STS) ，可從http://spring.io/tools進行下載.
對于客戶端，你需要.NET 3.5+ 框架.
如何做
要快速開始，你可以使用ＭVC Spring模板來創建一個完整簡單的Spring project，就像下面這樣:
1. 打開 STS，然后導航至File | New | Other | Spring Template Project |Spring MVC Project.
2. 修改POM.xml，并添加Tomcat的WebSocket libraries:
tomcat-coyote
tomcat-servlet-api
tomcat-catalina
3. 當然,也需要將RabbitMQ client library添加到POM中:
amqp-client
4. 創建一個RabbitMQ實例來存儲RabbitMQ參數
5.為了使用WebSocket，需要創建一個繼承自Tomcat WebSocketServlet類的RmqWebSocket類.在RmqWebSocket.java文件中，你也可以找到RabbitMQ消費者代碼.
6. 創建monitor_exchange_5_1 交換器，并在上面綁定一個消費者.
7. 實現ActualConsumer類中的handleDelivery方法，并將消息重定向到已連接的clients, 并按下面來迭代clients列表:
for(ClientMessage item: clients){
CharBuffer buffer = CharBuffer.wrap(message);
try {
item.myoutbound.writeTextMessage(buffer);
8. 在client端, 將使用JavaScript來連接WebSocket實例:
new WebSocket('ws://' + window.location.host + window.location.pathname+ "websocket");
9. 實現ws.onmessage()事件，并解析JSON消息來更新charts:
var obj = jQuery.parseJSON(message.data);
xcpu =obj.CPU ;
xmem = obj.MEM;
xupdate = obj.UPDATETIME;
var data = google.visualization.arrayToDataTable([['Label', 'Value'], ['Memory',xmem],['CPU', xcpu]]);
10. 作為一個生產者，我們創建一個.NET程序.一般說來，生產者需要將JSON消息發布到monitor_exchange_5_1交換器中。交換的JSON消息看起來像下面的代碼：
{"UPDATETIME":"23/06/2013 22:55:32","SERVERID":"1","CPU":10,"MEM":40}
到這里，程序就準備好了，但如果你想要配置Spring Insight的話,你還需要執行下面的步驟:
11. 轉到server section,然后點擊new server wizard, 選擇 vFabric server，如下所示:
12. 然后,移動并配置實例名稱和insight標志，如下所示:
13. 添加步驟１－１０中創建的web程序.
14. 啟動vFabric server，并轉向http://localhost:8080/insight/address.
15. 為了能快速安裝RabbitMQ Insight plugin,你可以從http://maven.springframework.org/release/com/springsource/insight/plugins/insight-plugin-rabbitmq-client/下載.
16. 將insight-plugin-rabbitmq-client-XXX.jar文件到你的$STSHOMEINSTALLATION/vfabric-tc-server-developer-2.8.2/mytcinstance/insight/collection-plugins文件夾中.
17. 最后，拷貝RabbitMQ Java client 到$STSHOMEINSTALLATION/vfabric-tcserver-developer-2.8.2/mytcinstance/insight/lib中.
18. 在你的瀏覽器指向http://localhost:8080/example來實時地獲取服務器負載信息，正如食譜開頭截屏中報告的一樣
19. 在你的瀏覽器中指向http://localhost:8080/insight或在 Insight 主頁點擊Click if insight is enabled! 按扭.
如何工作
在步驟1-3中，我們已經配置了環境;默認的向導模板會創建一個Maven project.
Spring 3.2.x 不支持WebSocket(Version 4.x支持),因此我們需要在POM.xml文件中添加websocket Tomcat依賴(步驟2).
RabbitMQ java client也存在于Maven repository, 在步驟３中，我們添加了此依賴.
RabbitMQinstance bean 包含RabbitMQ 連接參數.
bean的生命周期在root-context.xml文件中進行定義,帶有init和destroy方法和自定義參數. (可查看http://static.springsource.org/spring/
docs/3.2.x/spring-framework-reference/html/beans.html#beans-factorylifecycle來了解更多信息)
當client向monitor_exchange_5_1交換器發送消息時(步驟10)時,web程序會收到消息并將其重定向到WebSocket client的瀏覽器(步驟7).
當JSON消息到達瀏覽器的web頁面時，JQuery將對其進行解析，并最終用來更新Google Chart的餅圖.
TIP
你可以將程序部署到Tomcat.如果你使用STS標準服務器部署,應用程序URL就像下面這樣:
http://localhost:8080/example
步驟11-14 支持Spring Insight配置,因此你可以監控你的應用程序;使用RabbitMQ Insight plugin,你也可以監控RabbitMQ, 就像下面截圖展示的一樣:
更多
Spring Insight 可以幫你監控程序的行為.感謝這個插件,你也可以監控RabbitMQ. 這可能是較為復雜的主題，我們猜你從下面的連接讀取到的：
http://gopivotal.com/products/pivotal-tc-server
http://blogs.vmware.com/management/2013/01/new-video-deep-dive-intospring-insights-plugins-for-spring-integration-and-rabbitmq.html
無論如何,我們仍然會在后面的食譜中使用.
使用Spring來開發異步web搜索
當開發一個小型站點時，通常應用服務器會直接在關系數據庫上直接執行查詢，在這種情況下，它可能是非常快速的，且易于構建。當站點增長時，這種模式會出現伸縮性問題，特別是當服務器執行了多個數據查詢操作，再將最終結果呈現給終端用戶之前。另一個問題是系統升級，每個小升級都需要應用服務器停止和重新啟動。
在這個例子中，我們將展示如何使用消息隊列來伸縮web程序。我們將使用RabbitMQ來創建具有不同責任的單獨模塊。
傳統web應用程序模式(只存在兩個模塊:一個web server和一個數據庫server，它們是直接相連的)如下:
- 通信是異步的
- 每個模塊都不知道其它模塊的數目和網絡位置
添加、刪除和更新的模塊可以無縫地完成對網站的響應能力
- 不同scope的新模塊可以容易地添加
- 架構是可伸縮性的
相同的缺點如下:
- 程序過于復雜
- 本地化的，當與簡單快速方案比起來，程序較慢
在本例中，我們將實現一個異步的，可伸縮性的，平衡的搜索引擎來使用關鍵字來查詢一個條目(a book).
下面的圖解釋了我們即將構建的架構:
正如你看到的，我們將應用程序的職責做如下劃分:
- Tomcat只是接收HTTP請求和發布消息到RabbitMQ broker的代理
- Java 模塊以負載均衡的方式獲取消息，按需地執行數據庫查詢，并返回結果
- Tomcat獲取到結果后轉發給瀏覽器
TIP
為了簡化例子，為對每個java模塊模擬分布式數據庫，我們創建一個簡單的book列表，在這里，每個模塊都有整個數據的拷貝.在真實的場景中，你可能會有一個集群數據庫，或在分布式數據庫的前端有一個集群緩存.
與前面例子不同的是，在這里我們沒有使用WebSocket,而是使用Spring來返回結果: DeferredResult (http://static.springsource.org/spring/
docs/3.2.0.BUILD-SNAPSHOT/api/org/springframework/web/context/request/async/DeferredResult.html).
你可在Chapter05/Recipe02/web 找到源碼，客戶端代碼在Chpter05/Recipe01/backend.
準備
要使例子工作，你需要Java 1.7+ 和Apache Maven.
為了構建web程序,我們假設你使用Spring Tool Suite (STS),可從http://spring.io/tools下載.
如何做
我們會跳過前面食譜中已經介紹過的的創建MVC模板應用程序和RabbitMQ配置, 并會直接跳到相關的主題.
下面的步驟將會展示如何來實現web模塊:
1. 在RabbitMQInstance bean的Init()方法用于環境初始化－創建了search_exchange_05/02交換器和queue_consumer_search_05/02 隊列.
2. 配置消息超時時間:
Map<String, Object>args = new HashMap<String, Object>();
args.put("x-message-ttl", 4000);
channel.queueDeclare(Constants.queue, false, false,false, args);
3. 創建應用程序UUID (http://en.wikipedia.org/wiki/Universally_unique_identifier) ，并使用它來作為路由鍵，以讓RabbitMQ client來訂閱search_exchange_05/02交換器:
channel.queueBind(myQueue,Constants.exchange,UUIDAPPLICATION);
4. 在HomeController類中添加下面的方法:
publicDeferredResult<String>searchbook(@RequestParam String bookid) {..}
5. 在web.xml中使用<async-supported>true</asyncsupported>來啟用異步servlet.
6. 在ConcurrentHashMap<String,DeferredResult<String>>中注冊HTTP調用.
7. 使用bookkey參數來將消息發布到queue_consumer_search_05/02 隊列.
8. 當消息到達SearchResultConsumer, 將會產生一個新結果.
9. 在request map中找到 guidRequest,并使用setResult(message)來將結果設置為DeferredResult.
讓我們按下面的步驟，實現Java后端模塊來繼續完成食譜:
10. 使用channel.basicQos(1)來訂閱queue_consumer_search_05/02隊列.
11. 處理消息和執行搜索:
Book res= myDB.get(idx);
String jsonResult="";
..
jsonResult= jsonWriter.write(res);
12. 將結果發布到search_exchange_05/02交換器:
Builder bob = new Builder();
Map<String, Object> header = new HashMap<String,Object> ();
header.put("guidrequest", guidRequest);
bob.headers(header);
channel.basicPublish(Constants.exchange, RoutingKey,bob.build(), jsonResponse.getBytes());
如何工作
在創建環境之后(步驟1),web程序就準備就緒了. 我們使用queue_consumer_search_05/02來放置查詢結果，并使用search_exchange_05/02 交換器來獲取查詢結果.
在步驟2中,我們創建一個應用程序UUID，并將其作為search_exchange_05/02交換器的路由鍵，原因是你可能運行了多個能快速進行后端查詢的多個Tomcat 實例和應用程序 。唯一的UUIDs可用來區分每個模塊。
在步驟3中，我們使用DeferredResult實現了/searchbook URL 的GET handler. DeferredResult 類使用servlet3.0中的異步支持特性.(參考https://blogs.oracle.com/enterprisetechtips/entry/asynchronous_support_in_servlet_3 來了解更多信息)
為了啟用servlet3.0的異步支持，我們修改了web.xml file(步驟4).
使用DeferredResult類,你可以通過其它不同的線程來設置結果。在這種情況下，傳入的結果將會觸發一個回調.
讓我們看一下每個HTTP請求都發生了什么.
當瀏覽器執行請求時，如執行 /searchbook?bookkey=5, searchbook handler 會接受請求，并為其分配一個新的UUID.然后，它會創建DeferredResult<String>，并在請求參數發送到queue_consumer_search_05/02隊列后，將其放入到requests map (步驟6) .消息包含了request UUID和application UUID:
Builder bob = new Builder();
Map<String, Object> header = new HashMap<String,Object> ();
header.put("guidrequest", guidRequest);
bob.headers(header);
bob.correlationId(UUIDAPPLICATION);
注意，java后端會獲取消息(步驟11), 它會執行數據庫查詢(在我們的例子中，使用localDbEmulation進行的模擬)，并將查詢結果發布search_exchange_05/02交換器 (步驟12).
為了能夠正確地重定向消息, Java后端模塊將發回了request UUID，并使用correlationId header字段作為路由鍵.通過這種方式，我們可以確保響應消息能夠路由到發送請求的模塊。
TIP
在步驟9中,我們使用channel.basicQos(1)方法來創建了一個負載均衡請求(已在Distributing messages to many consumers recipe in Chapter 1, Working with AMQP看到過了).
如果你添加了多個Java后端模塊,它們會被自動地均衡負載。
因此，當消息到達web程序時，它包含下面的東西:
- request UUID
- JSON格式結果
SearchResultConsumer 將處理消息，并在request map中查詢request，同時它也會將結果設置到DeferredResult:
DeferredResult<String> deferredResult =requests.get(guidRequest.toString());
..
deferredResult.setResult(jsonResultSet);
結果最終會展現在瀏覽器中，request 也會從map中刪除.
requests.remove(guidRequest.toString());
TIP
web 和后端程序可通過Maven來編譯.使用這種方式，部署WAR和JAR文件將更加容易
在本場景中，為操作設置超時時間是很重要的，這樣可以防止不斷膨脹的請求,DeferredResult類的帶超時時間參數(單位毫秒)的構造器可以幫助我們：
new DeferredResult<String>(3000)
當請求超時時，必須從request map中刪除request,可以像下面這樣定義適當的回調來完成:
deferredResult.onTimeout(new Runnable() {
public void run() {
rmq.requests.remove(guidRequest);
}
為了防止隊列中消息膨脹,我們使用了x-message-ttl來配置消息的過期時間.如果隊列中的消息生存時間超過了4秒,該消息將會被刪除(參考Chapter 2, Going beyond the AMQP Standard).為了使例子簡單，我們只決定刪除長請求和膨脹消息—有效但過于激烈.在真實的程序中，超時應該以報警的方式進行處理.參考Chapter 2, Going beyond the AMQP Standard來了解隊列超時;對于DeferredResult, 需要注意超時處理器中的錯誤:
deferredResult.onTimeout(new Runnable() {
public void run() {
rmq.requests.remove(guidRequest);
}
TIP
超時監控是非常重要的，它可以作為擴展web應用程序的一個信號，也可以當作是出現問題的信號。
為了完成這個例子，我們使用了Apache JMeter (http://jmeter.apache.org/)來對website作壓力測試，以了解其能承受的負載。
在Chapter05/Recipe02/jm/ 文件夾, 你可以找到用來測試本食譜的JMeter文件,正如下面截屏看到的一樣:
你可以通過查看日志文件或STS控制臺來檢查程序行為是否正確.如果所有都是按預期執行的話，你會看到下面的日志信息:
INFO :com.test.bean.RabbitMQInstance - you have 0 pending requests
再次說明，你可以使用Ｍaven來編譯例子，然后部署web WAR file,并使用 java -cp rmq-0.0.1-SNAPSHOT.jar:./rabbitmq-client.jar com.test.
rmq.Main來執行后端jar文件. 這樣一來，你可以在瀏覽器打開 http://localhost:8080/rmq-1.0.0-BUILDSNAPSHOT/ ，并像下面一樣來執行搜索:
更多
在本食譜中，我們看到了隊列系統是企業系統的根基. 在此架構中，可以容易地添加多個Tomcat instances或更多的Java后端模塊來擴展程序。
新應用程序的集成是非常容易，因為可以將新程序非常容易地綁定到交換和隊列上.
同時這種架構對于管理軟件系統升級來說，也可以避免程序的停機時間.
也可參考
更多信息，可以閱讀下面的連接:
https://blogs.oracle.com/enterprisetechtips/entry/asynchronous_support_in_servlet_3
使用STOMP來開發web監控程序
在本食譜中，我們將展示如何使用Web-Stomp插件在web程序中來直接連接RabbitMQ. 我們將構建一個與Developing web monitoring applications with Spring 食譜一樣的例子.
STOMP 是簡單(或流)文本面向消息協議(http://stomp.github.io/),它是另一種消息傳輸協議，相對于AMQP來說，功能不強但非常輕量的客戶端。
Web-stomp RabbitMQ plugin 只提供了SockJS 服務器上的STOMP交互協議. On a different node, the STOMP plugin that we will see in Chapter 9, Extending
RabbitMQ Functionality provides just plain STOMP protocol interoperability.
你可在 Chapter05/Recipe03/HTML中找源碼.
準備
你只需要一個文本編輯器.
如何做
要烹飪此食譜, 我們將使用Web-Stomp (https://www.rabbitmq.com/web-stomp.html)插件, 它是模擬WebSockets的STOMP橋梁,并且我們會使用JavaScript STOMP client來進行訪問.
1. 啟用插件.
rabbitmq-plugins enable rabbitmq_web_stomp
2. 重啟RabbitMQ.
3. 通過http://127.0.0.1:15674/stomp來測試插件，正常的話，你會看到Welcome to SockJS!信息.
4. 創建一個簡單的HTML頁面，并增加下面的library:
<script src="http://cdn.sockjs.org/sockjs-0.3.min.js"></script>
5. 從 https://raw.github.com/jmesnil/stompwebsocket/master/dist/stomp.js下載Stomp.js，并將其拷貝到HTML文件附近js/stomp.js.
6. 在HTML頁面引入Stomp.js:
<script src="js/stomp.js"></script>
7. 初始始連接參數:
var ws = new SockJS('http://' + window.location.hostname +':15674/stomp');
var client = Stomp.over(ws);
8. 將client連接到RabbitMQ broker:
client.connect('guest', 'guest', on_connect, on_error, '/');
9. 讓client訂閱交換器:
var on_connect = function() {
client.subscribe("/exchange/monitor_exchange_05_01/stats",function(d) {
10. 解析消息并更新統計信息:
var obj = jQuery.parseJSON(d.body);
xcpu = obj.CPU;
xmem = obj.MEM;
xupdate = obj.UPDATETIME;
如何工作
一旦我們啟用并測試了插件(步驟1-3), 你就可以使用JavaScript頁面來連接RabbitMQ.第一個需要的包(步驟4)是Sockjs JavaScript library (你可在http://sockjs.org上找到更多的信息)，第二個需要的包是 Stomp.js. Stomp.js, 它是WebSocket上的STOMP JavaScript library (也可以參考http://jmesnil.net/stompwebsocket/doc/). 在初始化參數之后(步驟7), client 連接上RabbitMQ后，就可以on_connected事件上進行訂閱.
client使用subscribe 方法來訂閱 (/exchange/monitor_exchange_05_01/stats), 三個反斜杠隔開的子串分別表示：
- 第一個子串是交換器的類型或者是queue; 在我們的例子中，是交換器
- 第二個子串是名稱;在我們的例子中是monitor_exchange_05_01
- 第三個子串是路由鍵;在我們的例子中是stats
正如第一個食譜Developing web monitoring applications with Spring,你可以運行.NET client來向monitor_exchange_05_01交換器發送狀態信息.
這個例子中與第一食譜中的效果是一樣的，你可以將它們放到一起來執行.即使我們使用了不同技術，消息會同時到達這兩個web程序。
TIP
與第一個食譜不同的是，要部署HTML頁面,你只需要一個web服務器，如Apache.
本食譜中使用的是HTML5，因此須確保你的瀏覽器支持HTML5; 你也可以手機瀏覽器，就像下面截圖所示：
posted on 2016-06-14 22:07
胡小軍 閱讀(2284)
評論(0) 編輯 收藏 所屬分類:
RabbitMQ