node.js實現mqtt 發佈/訂閱 消息到主題

node.js實現mqtt 發佈/發送 消息到主題

---

• mqtt是啥?個人博客有寫這個東西:傳送門

安裝

首先你要安裝node.js和npm 教程傳送門

隨後找個文件夾執行 命令行 安裝mqtt模塊;

npm install mqtt

若是須要服務端 執行

npm install mosca

使用

• mqtt的文檔 如下代碼中有些參數能夠從mqtt官方文檔中獲得解釋
• node.js中mqtt模塊文檔
• mqtt協議入門文章

服務端

var mosca = require('mosca');
//構建自帶服務器
var MqttServer = new mosca.Server({
    port: 1883
});
//對服務器端口進行配置， 在此端口進行監聽
MqttServer.on('clientConnected', function(client) {
    //監聽鏈接
    console.log('client connected', client.id);
});
/**
 * 監聽MQTT主題消息
 **/
MqttServer.on('published', function(packet, client) {
    //當客戶端有鏈接發佈主題消息
    var topic = packet.topic;
    console.log(packet);
    switch (topic) {
        case 'test':
            console.log('message-publish', packet.payload.toString());
            //MQTT轉發主題消息
            MqttServer.publish({ topic: 'other', payload: 'sssss' });
            break;
        case 'other':
            console.log('message-123', packet.payload.toString());
            break;
    }
});

MqttServer.on('ready', function() {
    //當服務開啓時
    console.log('mqtt is running...');
});

PS: 不推薦node.js 做服務端

發佈消息到主題

var mqtt = require('mqtt');
var client = mqtt.connect('mqtt://127.0.0.1'); //鏈接到服務端
//client.subscribe('presence');
var num = 0;
var qtt = {}; //定義消息（能夠爲字符串、對象等）
qtt = 'setr=xxxxxxx1xx';
setInterval(function() { //一秒鐘發送一次 消息到主題 SN69143809293670state 消息爲 setr=xxxxxxx1xx
    client.publish('SN69143809293670state', qtt, { qos: 0, retain: true });  
}, 1000);

訂閱主題

var mqtt = require('mqtt');  
var client2 = mqtt.connect("mqtt://127.0.0.1:1883");   //指定服務端地址和端口
  
client2.subscribe('test',{qos:1});//訂閱主題爲test的消息  
  
client2.on('message',function(top,message) {  
    console.log(message.toString());  
});

製做接口(簡化)

我所接觸的這個物理設備是發送給他一個 控制 或者 查詢 請求到他訂閱的主題中(ctr),
而且它接收到控制信息,去執行.執行成功時 則會將 狀態 發送到另外一個 我訂閱的主題中 .
使用php實現比較麻煩,而且返回信息時有時無,使用node.js更方便,返回信息也更快;

物理環境 :centos7.2 64位,裝有mqtt服務代理端 和 node.js;

• 須要安裝http 和 express 以及 mqtt 模塊 npm install 模塊名
• node.js http模塊文檔

var mqtt = require('mqtt');
var express = require("express");
var app = express();
var hostName = '127.0.0.1'; //http服務的提供服務ip
var port = 8080;
var num = 1;
person = new Object();
person.firstname = "Bill";
app.all('*', function(req, res, next) {
    res.header("Access-Control-Allow-Origin", "*");
    res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
    res.header("Access-Control-Allow-Methods", "PUT,POST,GET,DELETE,OPTIONS");
    res.header("X-Powered-By", ' 3.2.1')
    res.header("Content-Type", "application/json;charset=utf-8");
    next();
});//json header頭

app.get("/zhinengjiaju/get", function(req, res) {
    //若是有get請求/zhinengjiaju/get則執行回調中的代碼(方便!)
    
    // console.log("請求url：", req.path)
    // console.log("請求參數：", req.query)
    req.setTimeout(200); 
    //設置請求創建200ms 就中斷接受請求,但仍是在接收到返回信息後返回給它
    var client = mqtt.connect('mqtt://127.0.0.1:1883', {
        username: 'username',
        password: 'password',
        clientId: 'ap' + num
    });
    //創建鏈接
    client.on('connect', function() {
        var sn = req.query.sn;
        var k = parseInt(req.query.k) - 1;
        // 127.0.0.1:8080/zhinengjiaju/get?sn=SN69143809293670&k=1&v=3&cmd=setr
        
        client.subscribe(sn + 'state', { qos: 1 });
        //開始訂閱
        
        if (req.query.cmd != 'setr') {
            m = req.query.cmd;
            if (req.query.cmd == 'qk') {
                m = 'setr=1111111111';
            }
            if (req.query.cmd == 'qg') {
                m = 'setr=0000000000';
            }

        } else {
            var m = ['x', 'x', 'x', 'x', 'x', 'x', 'x', 'x', 'x', 'x'];
            m[k] = req.query.v;;

            m = req.query.cmd + '=' + m.join('');
        }
        //一系列簡單接口處理

        client.publish(sn + 'ctr', m, { qos: 1, retain: true }); // 'Hello mqtt ' + (num++)
        //發送
        client.end();
        //發送完後當即結束這次和服務端創建的請求
    });
    client.on('message', function(topic, message) { 
        //訂閱信息一直在運行,若是有設備返回信息到主題,就執行此回調
        aaak(message.toString());
        //將值經過aaak函數傳遞給res.end返回給頁面數據;
        client.end();
        

    });

    function aaak(aaaa) {
        var objaaaa = JSON.parse(aaaa);
        // console.log(objaaaa);
        num++;
        client.end();
        res.end(aaaa);
    }

})

app.listen(port, hostName, function() {
    console.log(`服務器運行在http://${hostName}:${port}`);
});

謝謝支持,感受不錯打個賞;