MQTT，是：
- 輕量級的消息訂閱和發布（publish/subscribe）協議
- 建立在TCP/IP協議之上
IoT，internet of things，物聯網，MQTT在這方面應用較多。
官方網站：http://mqtt.org/
MQTT協議是針對如下情況設計的：
- M2M（Machine to Machine） communication，機器端到端通信，比如傳感器之間的數據通訊
- 因為是Machine to Machine，需要考慮：
- Machine，或者叫設備，比如溫度傳感器，硬件能力很弱，協議要考慮盡量小的資源消耗，比如計算能力和存儲等
- M2M可能是無線連接，網絡不穩定，帶寬也比較小
MQTT協議的架構，用一個示例說明。比如有1個溫度傳感器（1個Machine），2個小的顯示屏（2個Machine)，顯示屏要顯示溫度傳感器的溫度值。
可通過MQTT V3.1 Protocol Specification查閱詳細規范的細節。
顯示器需要先通過MQTT協議subscribe（訂閱）一個比如叫temperature的topic（主題）：
當溫度傳感器publish（發布）溫度數據，顯示器就可以收到了：
注：以上兩張圖，取自MQTT and CoAP, IoT Protocols
協議里還有2個主要的角色：
- client，客戶端
- broker，服務器端
它們是通過TCP/IP協議連接的。
因為MQTT是協議，所以不能拿來直接用的，就好比HTTP協議一樣。需要找實現這個協議的庫或者服務器來運行。
這里是官方的Server support。
我服務器端使用nodejs開發，因此選擇了：
MQTT.js最基本使用
安裝是很簡單的：
npm install mqtt MQTT.js實現的服務器端
代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
|
var mqtt = require('mqtt');
//{'topicName':[clientObj,clientObj ..]}
var subscribeTopics={};
//創建服務器對象
var server = mqtt.createServer(function(client) {
//建立連接時觸發
client.on(
'connect', function(packet) {
client.connack({returnCode:
0});
});
//客戶端發布主題時觸發
client.on(
'publish', function(packet) {
var topic=packet.topic;
var payload=packet.payload;
//如果沒有創建空的主題對應的client數組
if(subscribeTopics[topic]==null){
subscribeTopics[topic]=[];
}
else{
//遍歷該主題下全部client，並逐一發送消息
for(var i in subscribeTopics[topic]){
var client=subscribeTopics[topic][i];
client.publish({
topic: topic,
payload: payload
});
}
}
});
//當客戶端訂閱時觸發
client.on(
'subscribe', function(packet) {
var topic=packet.subscriptions[0].topic;
//如沒有，創建空的主題對應的client數組
if(subscribeTopics[topic]==null){
subscribeTopics[topic]=[];
}
//如果client數組中沒有當前client，加入
if(subscribeTopics[topic].indexOf(client)==-1){
subscribeTopics[topic].push(client);
}
});
client.on(
'pingreq', function(packet) {
client.pingresp();
});
client.on(
'disconnect', function(packet) {
//遍歷所有主題，檢查對應的數組中是否有當前client，從數組中刪除
for (var topic in subscribeTopics){
var index=subscribeTopics[topic].indexOf(client);
if(index>-1){
subscribeTopics[topic].splice(index,
1);
}
}
});
});
//監聽端口
server.listen(
1883);
|
這是一個最基本的服務器端，消息的存儲和查詢都需要自己編程處理。
比如你如果需要用redis保存和觸發數據，可參考這篇中文文章：node mqtt server (redis pub/sub)。
MQTT.js實現的客戶端
代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
var mqtt = require('mqtt');
client = mqtt.createClient(
1883, 'localhost');
client.subscribe(
'testMessage');
client.publish(
'testMessage', '發布測試信息');
client.on(
'message', function (topic, message) {
console.log(message);
client.end();
});
|
寫的很簡易，訂閱了主題，然后向相同主題發布消息，接收到消息后client停止。
使用Mosca
MQTT.js只是實現了最基礎的MQTT協議部分，對於服務器端的處理需要自己完成。
有關MQTT.js是否實現了MQTT server，詳細的說明，可參見MQTT Server: MQTT.js or Mosca?
正好，Mosca在MQTT基礎上實現了這些，它可以：
- 作為獨立運行的MQTT服務器運行
- 集成到nodejs程序里使用
安裝很簡單：
npm install mosca bunyan -g 作為獨立服務器運行
運行：
mosca -v | bunyan 然后，還可以用我上文的客戶端代碼運行測試。
集成在自己程序中使用
我考慮的后端持久化，是用MongoDB。Mosca另外幾個選項：
- Redis，缺點是更注重作為緩存，而不適合可靠持久化
- LevelUp，頭一次聽說，不打算做技術准備了，是用nodejs的包裝起來的LevelDB
- Memory，使用內存，估計默認的就是這個，不適合我使用的情況
首先要安裝mosca的庫：
npm install mosca 然后，在本機將mongodb運行起來，應該就可以執行下面的代碼了：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
var mosca = require('mosca')
var settings = {
port:
1883,
backend:{
type:
'mongo',
url:
'mongodb://localhost:27017/mqtt',
pubsubCollection:
'ascoltatori',
mongo: {}
},
persistence:{
factory: mosca.persistence.Mongo,
url:
"mongodb://localhost:27017/mosca"
}
};
var server = new mosca.Server(settings);
server.on(
'ready', function(){
console.log('Mosca server is up and running');
});
server.on(
'published', function(packet, client) {
console.log('Published', packet.payload);
});
|
直接運行作者文檔中的代碼會在多次運行客戶端后出現錯誤，我是參考了他2天前加上的示例代碼。
作者Matteo Collina生活在意大利的博洛尼亞，寫代碼很勤奮，這個項目更新很快，是不是說明這個方向（mqtt）很活躍呢？
作者也寫了個幻燈片，MQTT and Node.js
MQTT高級問題
keepalive和PING
從這篇文章MQTT協議筆記之連接和心跳：
心跳時間(Keep Alive timer)
以秒為單位，定義服務器端從客戶端接收消息的最大時間間隔。一般應用服務會在業務層次檢測客戶端網絡是否連接，不是TCP/IP協議層面的 心跳機制(比如開啟SOCKET的SO_KEEPALIVE選項)。 一般來講，在一個心跳間隔內，客戶端發送一個PINGREQ消息到服務器，服務器返回PINGRESP消息，完成一次心跳交互，繼而等待下一輪。若客戶端 沒有收到心跳反饋，會關閉掉TCP/IP端口連接，離線。 16位兩個字節，可看做一個無符號的short類型值。最大值，2^16-1 = 65535秒 = 18小時。最小值可以為0，表示客戶端不斷開。一般設為幾分鍾，比如微信心跳周期為300秒。
下面的代碼中我設置的是10秒：
|
1
2
3
4
5
6
7
8
9
10
11
|
var mqtt = require('mqtt');
var settings = {
keepalive:
10,
protocolId:
'MQIsdp',
protocolVersion:
3,
clientId:
'client-b',
clean:
false
}
client = mqtt.createClient(
1883, 'localhost',settings);
|
可以使用MQTT.js編寫簡單的服務器代碼，觀察到服務器端接收到PING請求，並發回PING響應：
|
1
2
3
4
|
client.
on('pingreq', function(packet) {
client.pingresp();
console.
log('pingreq & resp');
});
|
完整代碼上面已經貼過，另見Gist
QoS
QoS在MQTT中有（摘自MQ 遙測傳輸 (MQTT) V3.1 協議規范）：
- “至多一次”，消息發布完全依賴底層 TCP/IP 網絡。會發生消息丟失或重復。這一級別可用於如下情況，環境傳感器數據，丟失一次讀記錄無所謂，因為不久后還會有第二次發送。
- “至少一次”，確保消息到達，但消息重復可能會發生。
- “只有一次”，確保消息到達一次。這一級別可用於如下情況，在計費系統中，消息重復或丟失會導致不正確的結果。
MQTT.js只是支持了MQTT協議，並沒有支持QoS，也就是說，只支持最低級別的“至多一次”（QoS0）。
Mosca支持QoS0和1，但不支持2，見Add support QOS 2
接收離線消息
我在應用中的一個主要場景是，使用MQTT.js+Mosca做聊天服務器。
默認Mosca是不支持離線消息的，表現的現象是，如果是有人（client-a）先在主題上發布了消息：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
var mqtt = require('mqtt');
var settings = {
keepalive:
10,
protocolId:
'MQIsdp',
protocolVersion:
3,
clientId:
'client-a'
}
client = mqtt.createClient(
1883, 'localhost',settings);
client.publish(
'testMessage', '發布new測試信息0',{qos:1,retain: true});
client.publish(
'testMessage', '發布new測試信息1',{qos:1,retain: true});
client.publish(
'testMessage', '發布new測試信息2',{qos:1,retain: true});
client.publish(
'testMessage', '發布new測試信息3',{qos:1,retain: true});
setTimeout(
function(){
client.end();
},
1000);
|
那么另外一個人（client-b），隨后訂閱，僅能看到最后一條消息：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
var mqtt = require('mqtt');
var settings = {
keepalive:
10,
protocolId:
'MQIsdp',
protocolVersion:
3,
clientId:
'client-b'
}
client = mqtt.createClient(
1883, 'localhost',settings);
client.subscribe(
'testMessage',{qos:1},function(){
console.log('subscribe ok.');
});
client.on(
"message", function(topic, payload) {
console.log('message: '+payload);
});
|
運行結果類似這樣：
subscribe ok.
message: 發布new測試信息3
離線消息，需要以下幾點：
- 客戶端訂閱設置QoS=1
- 客戶端連接屬性
clean: false，作用是斷開連接重連的時候服務器端幫助恢復session，不需要再次訂閱
用代碼說明以下，先運行這段代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
var mqtt = require('mqtt');
var settings = {
keepalive:
10,
protocolId:
'MQIsdp',
protocolVersion:
3,
clientId:
'client-b',
clean:
false
}
client = mqtt.createClient(
1883, 'localhost',settings);
client.subscribe(
'testMessage',{qos:1},function(){
console.log('subscribe ok.');
client.end();
});
|
然后執行剛才發布多條消息的代碼。再執行下面的代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
var mqtt = require('mqtt');
var settings = {
keepalive:
10,
protocolId:
'MQIsdp',
protocolVersion:
3,
clientId:
'client-b',
clean:
false
}
client = mqtt.createClient(
1883, 'localhost',settings);
client.on(
"message", function(topic, payload) {
console.log('message: '+payload);
});
|
運行結果類似這樣：
message: 發布new測試信息1 message: 發布new測試信息3 message: 發布new測試信息2 message: 發布new測試信息0 收到消息的順序是亂的，為什么會這樣，其實很好理解，為了小型受限設備以及網絡不穩定的情況，消息是不好保證順序的。
解決辦法是發送的消息帶時間戳，接收后再做排序。
另外，擔心客戶端沒有做client.end()而非正常退出，那么再次連接是否能恢復session，測試了一下，注釋client.end()，沒有問題，正常收到多條離線消息。
SSL連接
Mosca支持SSL連接，可根據Nodejs TLS創建公鑰私鑰。
然后類似這樣啟動：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
var mosca = require('mosca')
var SECURE_KEY = __dirname + '/../../test/secure/tls-key.pem';
var SECURE_CERT = __dirname + '/../../test/secure/tls-cert.pem';
var settings = {
port:
8443,
logger: {
name:
"secureExample",
level:
40,
},
secure : {
keyPath: SECURE_KEY,
certPath: SECURE_CERT,
}
};
var server = new mosca.Server(settings);
server.on(
'ready', setup);
// fired when the mqtt server is ready
function setup() {
console.log('Mosca server is up and running')
}
|
這部分我沒有測試，直接轉自Mosca Encryption Support。
認證和授權
在Mosca Authentication提供了個簡易的命令行，可創建賬號用於認證並授權。
但是它不適合我的需求場景，我需要自己編寫認證和授權的邏輯。
雖然在作者官方網站上未找到，但在問題管理記錄中提交了這方面的支持：Authentication & Authorization。
有下面兩條支持，應該可以寫出自己的回調，並集成到Mosca中：
- add a callback to authorize a publish.
- add a callback to authorize a subscribe.
不過這塊沒有寫代碼，只是大致能確定。
性能問題
MQTT.js並不是完整解決方案，不需要考慮它的性能問題。
說一下Mosca，有一個這方面問題作者的答復，what about mosca’s performance，問問題的還是個中國人，我前面還引用了他的文章。作者基本意思是：
It basically depends on the RAM. On an AWS large instance it can reach 10k concurrent connections, with roughly 10k messages/second. 轉自：https://www.cnblogs.com/yfliufei/p/4386439.html