在本章中我們將覆蓋:
- 使用.NET client
- 通過MQTT綁定iPhone應用與RabbitMQ
- 在Andriod上使用消息來更新Google Maps
- 通過Andriod后端來發(fā)布消息
- 使用Qpid來交換RabbitMQ消息
- 使用Mosquitto來交換RabbitMQ消息
- 使用.NET clients來綁定WCF程序
介紹
在前面的章節(jié)中，我們已經(jīng)介紹了基本概念。現(xiàn)在，我們要使用這些概念來創(chuàng)建真實的應用程序。在這里我們不會講解AMQP新的概念，但你會看到RabbitMQ的一些典型用法，我們會讓RabbitMQ與其它技術(shù)進交互，以實現(xiàn)終端用戶為桌面和移動平臺上的應用程序。事實上RabbitMQ / AMQP與一般的消息傳遞技術(shù)一樣，其主要目的是分離分布式應用程序中的不同模塊，這樣使其很容易地混合多種技術(shù)，并可有廣泛的選擇。
使用.Net client
.NET是一個官方的RabbitMQ clients。在這個例子中，我們將使用它來創(chuàng)建一個訂閱者 。我們會使用Microsoft Windows Presentation Foundation (WPF)來創(chuàng)建我們第一個桌面程序。 可通過 http://msdn.microsoft.com/en-us/library/ms754130.aspx來了解更多信息.
我們會將.NET client綁定到第1章使用AMQP中廣播消息食譜中的Java dispatcher上。你可在Chapter04/Recipe01目錄中找到例子的源碼.
準備
你需要Visual Studio 2008或更新的版本, .NET framework 3.5 或 4以上版本,以及RabbitMQ .NET client包, RabbitMQ.Client.dll,這些你可在https://www.rabbitmq.com/dotnet.html下載.
如何做
為了實現(xiàn)使用RabbitMQ來消費消息的 .NET 程序，我們需要執(zhí)行下面的步驟:
1. 創(chuàng)建一個WPF應用程序，然后添加RabbitMQ.Client.dll引用.
2. 連接broker:
connection_factory.Uri = "amqp://guest:guest@" +edRabbitmqHost.Text + ":5672/";
3. 在myLastnews.fanout_01/06 交換器上綁定一個臨時隊列:
String myqueue = channel.QueueDeclare().QueueName;
channel.QueueBind(myqueue,news_exchange, "");
4. 啟動一個訂閱者線程:
sub = new Subscription(channel, myqueue, true);
StartSubscriberThread(sub);
5. 接收消息:
foreach (BasicDeliverEventArgs e in sub) {
...
Encoding.UTF8.GetString(e.Body);
}
6. 將消息寫入列表框:
Action<String> action = delegate(String value){lsnews.Items.Insert(0,value);};
Dispatcher.BeginInvoke(action,Encoding.UTF8.GetString(e.Body));
如何工作
client綁定到了myLastnews.fanout_01/06交換器上(已在第1章的使用AMQP的廣播消息食譜中創(chuàng)建).一旦將隊列與交換器進行了綁定(步驟3),我們將使用StartSubscriberThread(sub)來啟動一個新的線程,且方法會會執(zhí)行new Thread(() => InternalStartSubscriber(sub)) (步驟4). 收到的消息(步驟5)將會使用BeginInvoke()方法將其添加到列表框中.
為了試驗這個例子，你們執(zhí)行下面的操作:
- 從Chapter01/Recipe06/Java_6 中使用 java -cp ./ rabbitmq-client.jar rmqexample.Publish來運行publisher
- 運行 .NET client并連接broker ? 結(jié)果如下圖所示:
更多
AMQP的其中一個目標是集成不同開發(fā)語言，我們應該已經(jīng)學會了如何使用下面的語言來分發(fā)相同的消息：
- Java
- Python
- Ruby
- .NET
? 在后面的食譜中，我們將增加其它clients和技術(shù).
也可參考
更全面的文檔，請訪問https://www.rabbitmq.com/dotnet.html.
通過MQTT來綁定iphone app與RabbitMQ
MQTT (Message Queue Telemetry Transport，消息隊列遙測傳輸)是一種快速且輕量的開源協(xié)議. MQTT廣泛用于物聯(lián)網(wǎng)。更多信息，可查看http://mqtt.org/ 和http://en.wikipedia.org/wiki/Internet_of_Things.
在本例中，我們將通過使用Mosquitto(http://mosquitto.org/)包的MQTT創(chuàng)建一個app來綁定ios移動設(shè)備與RabbitMQ .
我們也將看到如何來接收由下面的程序發(fā)出的消息:
- 在第3章節(jié)管理RabbitMQ－通過瀏覽器來管理RabbitMQ食譜中提出的RabbitMQ web 管理應用程序來
- 在第1章節(jié)使用AMQP-使用topic交換器來處理消息食譜中介紹的publisher
你可在Chapter04/Recipe02目錄中找到此例的源碼.
TIP
通過MQTT,可以發(fā)送和接收消息,但它不是Apple Push 通知的替換物.
準備
跟隨本食譜，你需要:
- Apple Xcode IDE (https://developer.apple.com/xcode/)
- iOS開發(fā)包(https://developer.apple.com/library/ios/navigation/)
- Mosquitto C 包(https://bitbucket.org/oojah/mosquitto/src)
如何做
在本例中，我們需要啟用MQTT插件以允許MQTT clients連接,然后按下面的步驟來創(chuàng)建一個iOS應用程序：
1. 使用 rabbitmq-plugins enable rabbitmq_mqtt 命令來在RabbitMQ上安裝MQTT插件.
2. 重啟RabbitMQ.
3. 創(chuàng)建一個無ARC (Automatic Reference Counting)的新Xcode project，如下所示:
4. 下載Mosquitto C 包(https://bitbucket.org/oojah/mosquitto/src) ,將從它的lib目錄中拷貝 .h，.c 文件到示例樹的libmosquitto空目錄中.
5. 導入Moquitto C包，并將包添加到 project property Header search path.
6. 在你的ＭQTT接口文件中添加下面的include指令:
#include "mosquitto.h"
7. 要使用包來啟動，你必須使用mosquitto_lib_init()來初始化;
8. 創(chuàng)建一個適用于所有操作的struct mosquitto *mosq; 9. 使用mosq = mosquitto_new(NULL, true, self)來初始化struct;.
10. 使用下面的代碼來罪魁禍首RabbitMQ broker:
mosquitto_username_pw_set(mosq, NULL, NULL);
mosquitto_connect(mosq, cip, 1883, 60);
11. 使用下面的代碼來訂閱隊列:
mosquitto_subscribe(mosq, NULL, "tecnology.rabbitmq.ebook", 0);
12. 使用mosquitto_message_callback_set(mosq,on_message)來設(shè)置消息回調(diào);
13. 使用mosquitto_loop(mosq,1, 1)來開啟循環(huán)來接收服務端消息;
14. 你的app準備好后，你就可以發(fā)送消息了。首先，你可以使用web管理控制臺：切換到amq.topic交換器，并使用路由鍵technology.rabbitmq.ebook來發(fā)布消息.
15. 如果你想使用第１章節(jié)使用ＡＭＱＰ－使用topic交換器來處理消息路由食譜中的Java publisher，你需要使用路由鍵technology.rabbitmq.ebook將myTopicExchangeRoutingKey_01/08交換器與amq.topic 交換器進行綁定，正如我們在第２章節(jié)中超載ＡＭＱＰ標準－理解交換器到交換器食譜中看到的一樣（譯者注：綁定除了可發(fā)生在隊列與交換器外，也可以綁定交換器和交換器）。
16. 使用mosquitto_disconnect(mosq) 來斷開client,隨后mosquitto_destroy(mosq)將釋放struct.
17. 最后，使用mosquitto_lib_cleanup()來釋放包資源。
如何工作
從版本3.0開始, RabbitMQ支持MQTT協(xié)議. 你可以使用rabbitmq-plugins來啟用. Mosquitto是最流行的MQTT實現(xiàn)。在這個例子中，我們使用了它的C client library.通過此插件，使得AMQP和MQTT之間可相互通信!
在初始化步驟后以及步驟9-12后,client就已經(jīng)準備就緒了，這意味著iPhone已經(jīng)創(chuàng)建了一個綁定到amq.topic 交換器上的隊列，其路由鍵為rabbitmq.ebook.現(xiàn)在你可以將隊列視為一個普通的AMQP隊列.
現(xiàn)在打開web管理控制臺(the Managing RabbitMQ from a browser recipe in Chapter 3, Managing RabbitMQ)，并使用綁定的路由鍵發(fā)送一個test message，如下所示：
你的iPhone現(xiàn)在已經(jīng)收到了test message！
現(xiàn)在我們假設(shè)第１章節(jié)使用ＡＭＱＰ－使用topic交換器來處理消息路由食譜是一個龐大企業(yè)系統(tǒng)中的一部分，它是難于修改的。
你不能修改publisher和consumer的軟件,但你需要將"book's" 信息重定向到iPhone.
非常簡單!使用e-2-e擴展就可從管理控制臺進行交換器綁定(參考第２章節(jié)超越ＡＭＱＰ標準－理解交換器到交換器食譜),在Add binding from this exchange panel中設(shè)置正確的路由鍵，如下所示:
交換器已經(jīng)綁定了.現(xiàn)在我們可以執(zhí)行第１章節(jié)使用ＡＭＱＰ－使用topic交換器來處理消息路由食譜中的操作，看看會發(fā)生什么？你會看到消息被正確地路由了！
所有通過路由鍵technology.rabbitmq.ebook來發(fā)送的消息都將被重定向到amq.topic exchange，然后再到你的iPhone app.
更多
在這個例子中，為了演示RabbitMQ能應用于多種應用上下文，我們混合使用了前面章節(jié)中學過的概念以及一些新的技術(shù)。
不修改現(xiàn)有軟件，知道如何來添加或刪除新模塊，這非常重要.這很棒，不是嗎?
TIP
如果Mosquitto server不在LAN內(nèi),在連接mosquitto_connect(mosq, cip, 1883, 60 /*keepalive*/ )配置正確的keep-alive參數(shù)是非常重要的。
也可參考
MQTT： http://mqtt.org/.
你需要閱讀整個Mosquitto C client 文檔：http://mosquitto.org/api/files/mosquitto-h.html.
TIP
You can run this example even on a MQTT server as test.mosquitto.org with the routing-key #.
Other interesting links are http://www.eclipse.org/paho/ and http://mqtt.org/2011/08/mqtt-used-by-facebook-messenger.
在Andriod上使用消息來更新Google Maps
RabbitMQ與Android使用起來是相當簡單的; Dalvik虛擬機的字節(jié)碼完全兼容桌面Java版本, 且RabbitMQ JAR文件包含的API文件也是一樣的。
在本食譜中，我們打算策劃一個如何在移動應用程序中使用RabbitMQ的真實例子,特別是使用其它Andriod設(shè)備發(fā)送的AMQP消息來更新Google Maps上的位置。
但，本食譜中兩食譜項目中的第一部分,因此，在這里我們主要關(guān)注于應用程序消息，在下一個食譜中，我們將展示如何在Andriod設(shè)備中使用消息邏輯 。
兩個食譜也是互補的,在本食譜中，我們將在Anriod上展示RabbitMQ consumer，它將在地圖上繪畫出收到的位置.producer會在下個食譜中展示，它會通過RabbitMQ來發(fā)布位置. 你可以在Chapter04/Recip中找到本食譜中的源碼。
準備
要使本食譜正常工作，你需要:
- Android SDK tools (http://developer.android.com/sdk/index.html)
- Andriod API至少為11的Android設(shè)備(Honeycomb 和 Android 3.0.x或更高), 并提供OpenGL ES 2.0支持
- 需要在Andriod設(shè)備上安裝Google Play
如何做
1. 從Android SDK Manager中安裝Google Play Services:
2. 從Eclipse中,創(chuàng)建一個新的Android application. 然后跟隨向?qū)В钚〉?span style="background-color: inherit;">SDK必須是API11. 創(chuàng)建一個空的Activity.
3. 仔細參照Google Play服務提供的說明： http://developer.android.com/google/play-services/setup.html,
以及如何使用Google Map Android API v2上的說明： https://developers.google.com/maps/documentation/android/start.
4. 將RabbitMQ library files (commons-cli-1.1.jar, commons-io-1.2.jar,and rabbitmq-client.jar)拷貝到project的libs目錄.
5. 在由Eclipse向?qū)?chuàng)建的"Hello world!" TextView位置 ,插入一個片斷—你可以編輯對應的XML布局文件activity_main.xml，并插入下面的內(nèi)容：
<fragment
android:id="@+id/mapFragment"
android:name="com.google.android.gms.maps.MapFragment"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:layout_alignParentLeft="true"
android:layout_alignParentTop="true" />
6. 創(chuàng)建一個包含經(jīng)度位置和緯度位置簡單的MapCmd類.
7. 創(chuàng)建單例枚舉(enum)MapController,它實現(xiàn)了MapCmd類的線程安全FIFO隊列.
8. 創(chuàng)建一個用于異步消費RabbitMQ消息的ActualConsumer類，正好我們在第１章,使用AMQP中看到的一樣，并覆蓋其handleDelivery方法:
public void handleDelivery(
String consumerTag,
Envelope envelope,
BasicProperties properties,
byte[] body) throws java.io.IOException {
String message = new String(body);
MapController.INSTANCE.AddCmdFromJson(message);
}
9. 創(chuàng)建RabbitmqHandler類，并暴露其Connect()和Disconnect()方法，在其內(nèi)部，它將調(diào)用所有需要的RabbitMQ API，以使app異步消費消息。本類中自定義的連接參數(shù)將以常量的方式進行定義。
10. 創(chuàng)建UpdateMap類，其內(nèi)部是Runnable 接口的MainActivity實例, 它會定期地執(zhí)行繪畫操作:
public class UpdateMap implements Runnable {
GoogleMapmMap;
publicUpdateMap(GoogleMapmMap_) {
mMap = mMap_;
}
@Override
public void run() {
MapController.INSTANCE.ExecuteCmds(mMap);
updateHandler.postDelayed(this, UPDATE_INTERVAL);
}
};
11. 在最初由Eclipse創(chuàng)建的MainActivity類中,填充OnCreate() 方法：
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
mMap = ((MapFragment)
getFragmentManager().findFragmentById(R.id.mapFragment)).getMap();
mMap.setMapType(GoogleMap.MAP_TYPE_HYBRID);
updateMap = new UpdateMap(mMap);
updateHandler = new Handler();
updateHandler.postDelayed(updateMap, UPDATE_INTERVAL);
rabbitmqHandler = new RabbitmqHandler();
AsyncTask<Void, Void, Void>aconnect = new
AsyncTask<Void, Void, Void>() {
@Override
protected Void doInBackground(Void... params) {
try {
rabbitmqHandler.Connect();
} catch (IOException e) {
// TODO handle this
}
return null;
}
};
aconnect.execute((Void)null);
}
12. 最后對稱地實現(xiàn)OnDestroy()方法來清理連接和資源.
如何工作
為了要使用Google Maps, 則必須安裝Google Play，以使其服務對于 Android SDK開發(fā)，調(diào)試，部署。另外，我們必須在Google API console
(https://code.google.com/apis/console)上注冊過Google Maps。
不要忘記在AndroidManifest.xml中插入你自己的key. 在我們的例子中，我們留了一個占位符：
<meta-data
android:name="com.google.android.maps.v2.API_KEY"
android:value="SET-YOUR-GOOGLE-ANDROID-MAPS-KEY-HERE"/>
在我們講述了一個基礎(chǔ)Google Maps 程序后(步驟 1-5),我們已經(jīng)通過RabbitMQ中介，實現(xiàn)了從其它設(shè)備接收地球坐標的必要邏輯。
因為需要響應的安卓移動應用程序,必須強制性從一個不同的線程上執(zhí)行接收操作,而主線程則執(zhí)行一些繪圖和地圖相關(guān)的操作。這就是我們要實現(xiàn)一個線程安全MapCmd類的FIFO隊列的原因，并使其對于 MapController 單例可用; MapCmd 對象是通過RabbitMQ consumer線程，調(diào)用MapController.INSTANCE.AddCmdFromJson()方法塞入隊列的.
MapController FIFO 隊列消費者是Android程序的主線程. 我們創(chuàng)建了一個Handler對象來從main線程中周期性地調(diào)用updateMap.run()方法,并從隊列中消費MapCmd命令, 以及更新地圖顯示。
要完成這個應用程序，我們必須使用IP地址和RabbitMQ broker的憑證來定制RabbitmqHandler類，此類負責處理與RabbitMQ的連接以及消費者線程的創(chuàng)建。
在Andriod程序的的main activity 中，我們創(chuàng)建并初始化了RabbitmqHandler實例(步驟11), 但值得注意的是我們在AsyncTask匿名對象中已經(jīng)調(diào)用了它的Connect()方法; 并且強制不在主線程（包含RabbitMQ連接的線程）中執(zhí)行任何網(wǎng)絡同步操作.
當我們在 Andriod設(shè)備上啟動應用程序時，我們可以看到Google Maps會加載默認的世界地圖，在我們發(fā)送消息前，什么都不會發(fā)生。要測試這個例子，需要在web瀏覽器中打開RabbitMQ管理控制臺 (參考第3章節(jié)－管理RabbitmMQ中的從瀏覽器中管理RabbitMQ食譜); 我們應該能夠看到一個從Andriod設(shè)備連接的臨時隊列.我們可以直接向這個隊列發(fā)布消息，也可以將消息發(fā)布到amq.fanout交換器,并指定負載為JSON 格式,如下面的代碼所示:
{"lat":44.4937,"lon":11.3430}
當我們提交消息后，應用程序會聚焦在這里. 通過發(fā)送多個這個點附近的坐標, 我們將繪制出下面的屏幕截圖:
更多
在本食譜中，我們已經(jīng)展示了如何在Andriod設(shè)備上開發(fā)基于消息通信的分布式應用程序.為了簡單起見，我們省略了許多關(guān)于身份認證，負載均衡，安全的細節(jié)。
此外,這個例子只完成了項目的一半，在這里，我們只實現(xiàn)了消費者,以及從RabbitMQ控制臺來發(fā)送JSON格式的消息。
項目的后半部分將在下個食譜中展示，在你跳到下個章節(jié)之前，你需要注意地理坐標的數(shù)據(jù)可以是任意的－例如，它也可以來自嵌入式GPS設(shè)備。
TIP
對于嵌入式設(shè)備的移動設(shè)備，出于性能優(yōu)化，應該盡量使用MQTT，而不是使用native RabbitMQ AMQP協(xié)議,這一點，我們已經(jīng)在通過MQTT來綁定iPhone app與RabbitMQ食譜中看到過了.
從Andriod后端發(fā)送消息
在本食譜中，我們將完成前面食譜中開始的項目。我們將實現(xiàn)一個Android app來發(fā)布當前設(shè)備的位置給RabbitMQ. 當然，這個例子相對于前一個例子來說，是比較簡單的，因為我們不會在地圖上繪制位置，但我們?nèi)匀粫枰?/span>Google Play 服務，因為我們需要使用它的位置服務。同時，對于設(shè)備上可用的位置傳感器，它也提供了一個有用的抽象，以使應用程序能夠兼容廣泛的Andriod設(shè)備。
準備
要完成此食譜 ，我們需要:
- Android SDK 工具 (http://developer.android.com/sdk/index.html)
- 一臺帶Android API 11級以上（Honeycomb 和Android 3.0.x 或更高)的Andriod設(shè)備
- Andriod設(shè)備上安裝有Google Play服務
如何做
1. 從Android SDK Manager中安裝Google Play 服務，這已經(jīng)在前面的食譜中看過了.
2. 創(chuàng)建 一個新的Android應用程序.跟隨向?qū)?需要的最小SDK必須是API11.創(chuàng)建一個空的Activity.
3. 仔細遵循Google的說明來設(shè)置Google Play服務(http://developer.android.com/google/play-services/setup.html).
4. 拷貝RabbitMQ library 文件(commons-cli-1.1.jar, commons-io-1.2.jar,and rabbitmq-client.jar)到項目的libs目錄中.
5. 在由 Eclipse wizard創(chuàng)建Hello world! TextView的地方, 插入一個名為followmeSwitch的 switch widget.你可以編輯相應的XML layout文件activity_main.xml，并插入下面的代碼:
<Switch
android:id="@+id/followmeSwitch"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_marginLeft="28dp"
android:text="@string/followme" />
6. 創(chuàng)建一個繼承至IntentService的SenderService類.
7. 在MainActivity.java中設(shè)置開關(guān)動作:
final Switch fms =(Switch)findViewById(R.id.followmeSwitch);
fms.setOnCheckedChangeListener(new CompoundButton.OnCheckedChangeListener() {
@Override
public void onCheckedChanged(CompoundButtonbuttonView,
booleanisChecked) {
if (isChecked) {
serviceIntent = new Intent(MainActivity.this,
SenderService.class);
startService(serviceIntent);
} else {
stopService(serviceIntent);
} }
});
如何工作
在本食譜中，我們接近了混合使用RabbitMQ和Android技術(shù)的另一種模式: 在后端來發(fā)布消息，即時是用戶界面在Andriod設(shè)備上不可視。
首先，我們創(chuàng)建了一個帶開關(guān)的標準Android應用程序：當打開時，我們可以向RabbitMQ broker來分發(fā)消息，當關(guān)閉時，我們可以停止消息的分發(fā)。
因此，我們將followmeSwitch操作綁定了SenderService中 (步驟7).記住這點是很重要的：當我們在Andriod服務中啟動一個后臺job時，后臺事件依然是通過應用程序的主線程來執(zhí)行的，并且我們也需要采用與設(shè)計Andriod GUIS時相同的響應方針.
但在我們的例子中，我們已經(jīng)決定使用IntentService 幫助類來啟動一個不間斷運行的后端線程.在這個線程中，我們將連接broker,并且通過調(diào)用locationClient.getLastLocation()方法周期性地發(fā)送設(shè)備當前位置給RabbitMQ(參考步驟6中提到的源文件代碼).
TIP
在這個例子中，我們使用了Andriod標準的序列化類android.util.JsonWriter來代替前面我們看到的RabbitMQ JSONWriter. 事實上，后者不能在Andriod設(shè)備上工作，因為Dalvik只包含了Java beans的子集，缺少了相關(guān)的反射.
只要用戶關(guān)閉了開關(guān)，我們就會立即退出線程.通過在一臺設(shè)備上運行此程序，并在另一臺設(shè)備上運行前面食譜中的程序，我們可以通過RabbitMQ消息來交流我們的地理位置。
更多
Google localization client library允許位置變化時觸發(fā)事件,因此這應該是發(fā)送地理位置坐標的最佳方法。這種方案可以優(yōu)化電池的使用，但為了簡單起見，我們沒有這樣做。但就此應用程序來說，功能是完整的。你可以運行多個不同的設(shè)備來允當消費者－當我們將消息發(fā)送到fanout時，它們會同步的更新位置。你甚至可以并發(fā)地運行多個生產(chǎn)者，但當前實現(xiàn)只是將它們的數(shù)據(jù)混合在一起，如果要從多個不同設(shè)備來繪制多個不同軌跡，你需要替換fanout交換器，每個生產(chǎn)者需要使用topic交換器且?guī)Р煌穆酚涉I來發(fā)送消息。
使用Qpid來交換RabbitMQ消息
在本食譜中，我們將看到如何使用RabbitMQ client來與Apache Qpid交互.
準備
你需要從http://qpid.apache.org/download.html下載 Qpid;我們使用的是Java版本.
如何做
1. 從前面的章節(jié)中來獲取Java示例,如：第１章使用AMQP中的廣播消息或使用topic交換器來處理消息路由食譜中的例子。
2. 運行Qpid server.
3. 運行clients.
如何工作
當你從第１章使用AMQP中的廣播消息中的例子來執(zhí)行client的時候，Qpid server與的RabbitMQ的行為一樣，如下所示：
TIP
QPid 使用相同的RabbitMQ端口,因此如果要在同一臺機上運行這兩者，你可以修改端口或者關(guān)閉RabbitMQ（端口沖突）.
使用Mosquitto來交換RabbitMQ消息
在這個例子中，我們會創(chuàng)建一個java proxy，此代理會獲取來自Andriod后臺來發(fā)布消息食譜例子中發(fā)布的消息，并將它們轉(zhuǎn)發(fā)到任何一個移動OS,正如我們在通過MQTT來綁定iPhone應用程序到RabbitMQ食譜中看到的一樣，可按下面的截圖來使用MQTT服務器:
準備
你需要Paho Java library client, 你可在http://git.eclipse.org/c/paho/org.eclipse.paho.mqtt.java.git/中找到.
如何做
1. 你需要創(chuàng)建一個amq.fanout交換器的消費者:
String myQueue = channel.queueDeclare().getQueue();
String myExchange = "amq.fanout";
channel.queueBind(myQueue,myExchange,"");
2. 使用Paho Java client來創(chuàng)建一個MQTT連接:
mqttClient = new MqttClient(MQTTip,
MqttClient.generateClientId());
mqttClient.connect();
3. 將獲取到的GPS消息發(fā)送給MQTT服務器:
mqttClient.publish("follow.me", message, 0, false);
4. 當通過MQTT綁定iPhone app與RabbitMQ時，修改路由的topic:
mosquitto_subscribe(mosq, NULL, "follow.me", 0);
5. 當通過MQTT綁定iPhone app與RabbitMQ時，使用MQTT服務器IP的地址修改服務器的IP
如何工作
按照步驟１－３，我們已經(jīng)創(chuàng)建了兩個連接,一個是RabbitMQ，一個是MQTT服務器。
當Android app發(fā)送了一些消息時，消息者接收到消息后會使用mqttClient.publish("follow.me", message,0,false)將其發(fā)布Mosquitto server;.
在iPhone程序上，你需要使用follow.me來修改topic,你也需要修改服務器IP地址，結(jié)果如下面的屏幕截圖所示：
因此,如果Android app啟用了后，你可以在你的iPhone app中看經(jīng)度和緯度.
TIP
我們使用了公共服務器test.mosquitto.org，在使用前，請先網(wǎng)站的免責聲明.
更多
We have divided the broker's responsibility, one for the AMQP protocol and one for the MQTT protocol.
也可參考
你可在http://www.eclipse.org/paho/找到更多關(guān)于Paho的更多信息. 當你創(chuàng)建Paho Java client 源碼時，構(gòu)建器會在你的文件系統(tǒng):
yourbasedir/org.eclipse.paho.mqtt.java/org.eclipse.paho.client.mqttv3/build/dist/javadoc中創(chuàng)建javadoc.
使用.Net來綁定WFC應用程序
在這個例子中，我們將展示如何使用不同的技術(shù)來構(gòu)建一個分布式應用程序. 其目標是創(chuàng)建一個健康程序來收集本地網(wǎng)絡應用程序的狀態(tài)信息。如果出現(xiàn)問題的話，應用程序會觸發(fā)警報并將其發(fā)送給移動設(shè)備，在這里，我們使用Amazon EC2云來提供高可用性。但，EC2不是強制性的.
我們將使用Windows Communication Foundation (WCF) 和 RabbitMQ WCF 服務模塊來創(chuàng)建這個例子. WCF service 將扮演主機健康收集器，并且會有多個客戶端會向WCF服務發(fā)送消息。
RabbitMQ service 模塊允許在AMQP之上使用SOAP協(xié)議. 我們也會將消息重定向到 also redirect the messages to a mobile iOS application via MQTT using the proxy already seen in the recipe
Exchanging RabbitMQ messages with Mosquitto.
In this recipe, we are going to use AMQP, SOAP, and MQTT as protocols, RabbitMQ, WCF,and iOS as technologies, and finally, an AWS EC2 instance as the server. The architecture is
summarized in the following figure:
準備
你需要Visual Studio 2005或2008, .NET framework 3.5,以及RabbitMQ WCF 服務模塊, 這些你可以在http://www.rabbitmq.com/dotnet.html中找到.
下載WCF library的zip包.
如何做
1. 創(chuàng)建一個WCF應用程序,然后通過拷貝project中l(wèi)ibs目錄下的RabbitMQ.ServiceModel.dll來配置RabbitMQ service模塊. 接著通過右部按扭來打開App.config文件，并導航至Edit WCF configuration | Advanced | binding extension | new 并選擇剛剛拷貝的文件。
2. 通過導航至WCF configuration | bindings | new bind configuration來創(chuàng)建一個綁定, 并按下面的截圖為配置RabbitMQ實例：
3. 通過導航至Edit configuration | Endpoints | New Service End point來配置WCF end point, 然后設(shè)置Contract為WcfLib.IService1, rabbitmqExtension的Binding屬性 (在步驟１中已配置), 以及rabbitmqBinding的BindingConfiguration屬性(步驟2已配置). 其結(jié)果如下所示：
4. 導航至Edit configuration | services | endpoint | new and insert soap.amqp:///來配置host.
5. 創(chuàng)建Service1接口方法SendMachineInfo(MachineInfomachine Info),并在Service1 類中實現(xiàn).
6. 創(chuàng)建一個新的C# project，為了使用接口方法，還需要添加服務端依賴.
7. 配置client App.conf ，并為RabbitMQ service 模塊添加配置 (步驟 1和2 ), 然后導航至Edit configuration | Client | endpoints and adding soap.amqp:///Service1來配置client service; 綁定截圖如下：
8. 使用ClientService :ClientBase<IService1>, IService1來啟動client .
9. 使用base.Channel.SendMachineInfo(machineInfo)來調(diào)用接口;
10. 如果你喜歡，你可以修改Exchanging RabbitMQ messages with Mosquitto 食譜來向iphone發(fā)送警報.你只需要修改下面的行：
String myExchange = "amq.direct";
channel.queueBind(myQueue,myExchange,"/Service1");
如何工作
根據(jù)步驟1-4, 我們使用RabbitMQ繼承了WCF 標準綁定, 并且我們創(chuàng)建了一個單向通信，它是一種發(fā)后即忘模式,且服務端只創(chuàng)建了一個綁定至 amq.direct 交換器的隊列，其路由鍵為/Service1.
MachineInfo類是按照SOAP來編碼的 (步驟9); 服務器隨后會執(zhí)行函數(shù)SendMachineInfo(MachineInfomachineInfo). SOAP消息作為AMQP消息的內(nèi)嵌負載。在這個例子中，服務端將警告消息存儲在C:\warningstatus.store文件中,clients每秒都會檢查CPU和內(nèi)存的使用情況，并在出現(xiàn)警告時執(zhí)行遠程調(diào)用。如果你執(zhí)行步驟１０，你就會在app中收到消息，就像下面展示的一樣：
Mosqui
更多
在這個例子中，我們已經(jīng)嘗試使用了RabbitMQ Amazon Elastic Compute Cloud(EC2),以及不同網(wǎng)絡中的WCF WCF客戶機和服務器。理解RPC調(diào)用是由broker來保證，且與標準RPC調(diào)用（模塊之間直接通信）有所區(qū)別，這是很重要的。在這種方式中，架構(gòu)是松耦合的，RabbitMQ是當作bus來使用的，因此它可以附著不同的clients.WCF 綁定是一種Microsoft架構(gòu)與其它系統(tǒng)集成時的強有力工具。
你可以使用web管理控制臺或命令行來監(jiān)控隊列。
通信也可以是雙向的，使用兩個隊列:一個用于請求，另一個用于異步響應
也可參考
查看https://www.rabbitmq.com/dotnet.html 并找到 section A description of the WCF binding and Service Model.
查看http://aws.amazon.com/ec2/ 來了解更多關(guān)于Amazon EC2的信息;總之,我們也會在接下來的Amazon cloud章節(jié)中介紹.
posted on 2016-06-13 20:25
胡小軍 閱讀(1768)
評論(2) 編輯 收藏 所屬分類:
RabbitMQ