前言

前些日子了解到mqtt這樣一個協議，可以在web上達到即時通訊的效果，但網上并不能很方便地找到一篇目前版本的在node下正確實現這個協議的博客。

自己搗鼓了一段時間，理解不深刻，但也算是基本能夠達到使用目的。

本文尚未對離線消息的接收順序進行處理。

代碼

服務端: server.js

//服務端引入中間件moscalet mosca = require('mosca')let settings = { port: 5112}let server = new mosca.Server(settings)server.on('ready', function(){  console.log('Mosca server is up and running at port 5112'); })server.on('published', function(packet, client) { console.log('Published', packet.payload)})server.on('clientDisconnected', function(client){ console.log('disconnected: ', client.id)})

推送端: pub.js

//客戶端引入mqttlet mqtt = require('mqtt');let client = mqtt.connect('mqtt://localhost', { port: 5112, clientId: 'cli_pub',})let num = 0;setInterval(function (){ client.publish('test',  'Hello mqtt ' + (++num), {qos:1}, () => console.log(num));}, 1000)

訂閱端: sub.js

let mqtt = require('mqtt')let client = mqtt.connect('mqtt://localhost', { port: 5112, clientId: 'cli_sub',})client.subscribe('test',{qos:1})client.on('message', function (topic, message) { console.log('received message: ', message.toString())})

server運行后，先啟動pub，再啟動sub，即可在sub中接收到推送過來的消息序列

至此實現了簡單的即時推送

離線推送相關配置及簡要介紹

離線配置-服務端：

要實現消息的離線推送，必然需要一個存儲臨時數據的部件

此處用到的是mongo，當然可以根據需要選擇其他的存儲工具

server.js中的settings需更改為:

let settings = { port: 5112, persistence:{  //增加了此項  factory: mosca.persistence.Mongo,  url: "mongodb://localhost:27017/mosca" }}

factory: 引入mosca對特定存儲工具的一些處理方法

url: 其中的 27017 為mongo所監聽的端口號，mosca為存儲相關數據的數據庫

值得一提的是：配置好mongo的環境后，不需要提前在mongo中手動創建，若數據庫不存在會自動生成，而且mosca會為你作好其他一切基本事項 (即：若只想臨時體驗下效果，甚至可以暫時把mongo放一邊 )

在mongo中，可以看到自動新添了db: mosca及其下的collection(相當于關系型數據庫中的表/關系)

離線配置-客戶端：

pub.js和sub.js中的client中都可以改為：

let client = mqtt.connect('mqtt://localhost', { port: 5112, clientId: 'cli_**', clean: false//增加了此項})

• clientId: 區分客戶端的識別碼
• clean: 此處決定了客戶端在服務端的session是否會被清除，默認為true，為實現離線推送，我們需要將其保留
• clean及上文中的persistence為實現離線推送的關鍵配置

mqtt.connect()會返回一個mqttClient對象，包含了：reconnect(), subscribe(), publish()等一系列方法。

本文中發送端接收端被分為了pub.js和sub.js兩個獨立文件，僅僅為了方便在不同控制臺中觀察效果
一個client可以既為推送端，又為訂閱端

至此，所有代碼已完成

其他介紹：

client.subscribe():
為本客戶端訂閱一個話題，所有訂閱此話題的用戶都會收到在此話題下推送的信息

//client.subscribe(topic,opts)client.subscribe('test',{qos:1})

opts中的qos為通信機制，控制發送端與接收端的互鎖程度

上文中的其中一個collection: subscriptions即記錄各用戶話題訂閱情況

用戶cli_sub及cli2_sub訂閱了話題test:

(新增一個cli2_pub，下文有用)

注：

重復執行腳本sub.js實際上對topic進行了重復訂閱

實際編碼時，應避免topic的重復訂閱，即使重復訂閱并不影響實現效果

client.publish():

向指定topic發送數據

message為Buffer或String格式，可以通過序列化或轉json實現對復雜數據對象的傳送

//client.publish(topic, message, opts, callback)let num = 0;setInterval(function (){ client.publish('test',  'Hello mqtt ' + (++num), {qos:1}, () => console.log(num));}, 1000)

參數不再贅述

此處用一個定時器定時在 topic: test 下發送'Hello mqtt 1,2,3..'

用回調函數實時打印一下發送的num：

當訂閱者處于離線狀態時，可以在collection: packets中查看到臨時數據的存儲情況：

mosca把每一條推送消息為所有訂閱用戶都生成了獨立的記錄，用同一個messageId進行關聯

當其中一個用戶(cli2_sub)上線時，獲取到其對應的數據，

而后數據庫中相應記錄便會被刪除

此時僅有cli_sub用戶的數據

當cli2_sub上線接收消息后，packets中記錄將被清空

client.on():

即在client上觸發的事件，此處只列舉消息接收事件

//client.on(event, callback)client.on('message', function (topic, message) { console.log('received message: ', message.toString())})

處理為簡單地打印到控制臺

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。