node.js实现mqtt 发布/发送 消息到主题

1
var mosca = require('mosca');
2
//构建自带服务器
3
var MqttServer = new mosca.Server({
4
    port: 1883
5
});
6
//对服务器端口进行配置， 在此端口进行监听
7
MqttServer.on('clientConnected', function(client) {
8
    //监听连接
9
    console.log('client connected', client.id);
10
});
11
/**
12
 * 监听MQTT主题消息
13
 **/
14
MqttServer.on('published', function(packet, client) {
15
    //当客户端有连接发布主题消息
16
    var topic = packet.topic;
17
    console.log(packet);
18
    switch (topic) {
19
        case 'test':
20
            console.log('message-publish', packet.payload.toString());
21
            //MQTT转发主题消息
22
            MqttServer.publish({ topic: 'other', payload: 'sssss' });
23
            break;
24
        case 'other':
25
            console.log('message-123', packet.payload.toString());
26
            break;
27
    }
28
});
29

30
MqttServer.on('ready', function() {
31
    //当服务开启时
32
    console.log('mqtt is running...');
33
});

1
var mqtt = require('mqtt');
2
var client = mqtt.connect('mqtt://127.0.0.1'); //连接到服务端
3
//client.subscribe('presence');
4
var num = 0;
5
var qtt = {}; //定义消息（可以为字符串、对象等）
6
qtt = 'setr=xxxxxxx1xx';
7
setInterval(function() { //一秒钟发送一次 消息到主题 SN69143809293670state 消息为 setr=xxxxxxx1xx
8
    client.publish('SN69143809293670state', qtt, { qos: 0, retain: true });
9
}, 1000);

1
var mqtt = require('mqtt');
2
var client2 = mqtt.connect("mqtt://127.0.0.1:1883");   //指定服务端地址和端口
3

4
client2.subscribe('test',{qos:1});//订阅主题为test的消息
5

6
client2.on('message',function(top,message) {
7
    console.log(message.toString());
8
});

1
var mqtt = require('mqtt');
2
var express = require("express");
3
var app = express();
4
var hostName = '127.0.0.1'; //http服务的提供服务ip
5
var port = 8080;
6
var num = 1;
7
person = new Object();
8
person.firstname = "Bill";
9
app.all('*', function(req, res, next) {
10
    res.header("Access-Control-Allow-Origin", "*");
11
    res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
12
    res.header("Access-Control-Allow-Methods", "PUT,POST,GET,DELETE,OPTIONS");
13
    res.header("X-Powered-By", ' 3.2.1')
14
    res.header("Content-Type", "application/json;charset=utf-8");
15
    next();
16
});//json header头
17

18
app.get("/zhinengjiaju/get", function(req, res) {
19
    //如果有get请求/zhinengjiaju/get则执行回调中的代码(方便!)
20

21
    // console.log("请求url：", req.path)
22
    // console.log("请求参数：", req.query)
23
    req.setTimeout(200);
24
    //设置请求建立200ms 就中断接受请求,但还是在接收到返回信息后返回给它
25
    var client = mqtt.connect('mqtt://127.0.0.1:1883', {
26
        username: 'username',
27
        password: 'password',
28
        clientId: 'ap' + num
29
    });
30
    //建立连接
31
    client.on('connect', function() {
32
        var sn = req.query.sn;
33
        var k = parseInt(req.query.k) - 1;
34
        // 127.0.0.1:8080/zhinengjiaju/get?sn=SN69143809293670&k=1&v=3&cmd=setr
35

36
        client.subscribe(sn + 'state', { qos: 1 });
37
        //开始订阅
38

39
        if (req.query.cmd != 'setr') {
40
            m = req.query.cmd;
41
            if (req.query.cmd == 'qk') {
42
                m = 'setr=1111111111';
43
            }
44
            if (req.query.cmd == 'qg') {
45
                m = 'setr=0000000000';
46
            }
47

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

52
            m = req.query.cmd + '=' + m.join('');
53
        }
54
        //一系列简单接口处理
55

56
        client.publish(sn + 'ctr', m, { qos: 1, retain: true }); // 'Hello mqtt ' + (num++)
57
        //发送
58
        client.end();
59
        //发送完后立即结束此次和服务端建立的请求
60
    });
61
    client.on('message', function(topic, message) {
62
        //订阅信息一直在运行,如果有设备返回信息到主题,就执行此回调
63
        aaak(message.toString());
64
        //将值通过aaak函数传递给res.end返回给页面数据;
65
        client.end();
66

67

68
    });
69

70
    function aaak(aaaa) {
71
        var objaaaa = JSON.parse(aaaa);
72
        // console.log(objaaaa);
73
        num++;
74
        client.end();
75
        res.end(aaaa);
76
    }
77

78
})
79

80
app.listen(port, hostName, function() {
81
    console.log(`服务器运行在http://${hostName}:${port}`);
82
});