mqtt客户端工具：mqttx  服务端：emqx

js实现简单的mqtt发布订阅：

var mqtt = require('mqtt');
var client = mqtt.connect('mqtt://:1883', {
username: process.env.TOKEN
});

client.on('connect', function() {
console.log('connected');
client.subscribe('v1/devices/me/rpc/request/+')
});

client.on('message', function(topic, message) {
console.log('request.topic: ' + topic);
console.log('request.body: ' + message.toString());
var requestId = topic.slice('v1/devices/me/rpc/request/'.length);
var msg = JSON.parse(message);
console.warn(msg);
var respmsg;
if (msg["method"] == "get_properties") {
respmsg = {"result": []};
msg["params"].forEach(v => {
respmsg.result.push({"siid": v.siid, "piid": v.piid, "value": "image.png", "code": 0})
});
}else if (msg["method"] == "set_properties") {
respmsg = {"result": []};
msg["params"].forEach(v => {
respmsg.result.push({"siid": v.siid, "piid": v.piid, "code": 0})
});
}else if (msg["method"] == "action") {
respmsg = {"result": {"code":0, "out": []}};
respmsg.result.code = 0;
msg["params"]["in"].forEach(v => {
respmsg.result.out.push({"piid": v.piid, "value": v.value})
});
}
console.warn(respmsg);
//client acts as an echo service
client.publish('v1/devices/me/rpc/response/' + requestId, JSON.stringify(respmsg));
});