物联网宠儿mqtt.js那些事儿 #204
Labels
Comments
什么是微消息队列？消息队列一般分为两种：
目前我实践过的，也就是我们本篇博文深入分析的，是物联网消息队列的mqtt.js。 传统的消息队列（微服务间信息传递）传统的微服务间（多个子系统服务端间）消息队列是一种非常常见的服务端间消息传递的方式。 典型代表有：RabbitMQ，Kafka，RocketMQ。 更多微服务消息队列可查看：node-mq-tutorial 使用场景多种多样：
MQTT消息队列（物联网端与云间消息传递）MQTT是一个物联网MQTT协议，主要解决的是物联网IoT网络情况复杂的问题。 阿里云有MQTT消息队列服务。通信协议支持MQTT，STOMP，GB-808等。数据传输层支持TCP长连接、SSL加密、Websocket等。 使用场景主要为数据传输：
目前我手上负责的运行了2年的聊天系统就是使用的这个服务，我们主要按照 |
MQTT关键名词解释实例（Instance）每个MQTT实例都对应一个全局唯一的服务接入点。 客户端Id（Client ID）MQTT的Client ID是每个客户端的唯一标识，要求全局都是唯一的，使用同一个Client ID连接会被拒绝。 let CID_PC = `<GroupID>@@@-PC<DeviceID>`
let CID_Android = `<GroupID>@@@-Android<DeviceID>`
let CID_IOS = `<GroupID>@@@-IOS<DeviceID>`组Id（Group ID）用于指定一组逻辑功能完全一致的节点公用的组名，代表的是一类相同功能的设备。 Device ID每个设备独一无二的标识。这个需要保证全局唯一，可以是每个传感器设备的序列号，可以是登录PC的userId。 父主题（Parent Topic）MQTT协议基于Pub/Sub模型，任何消息都属于一个Topic。 子主题（Subtopic）MQTT可以有二级Topic，也可以有三级Topic。 |
P2P消息和Pub/Sub消息Pub/Sub消息就是订阅和发布的模式，类似事件监听和广播。 什么是P2P消息？
P2P模式和Pub/Sub模式的区别发送消息时
接收消息时
nodejs发送P2P消息const p2pTopic =topic+"/p2p/GID_xxxx@@@DEVICEID_001";
mqtt.client.publish(p2pTopic); |
封装的mqtt.js通用class
import mqtt from 'mqtt';
import config from '@/config';
export default class MQTT {
constructor(options) {
this.name = options.name;
this.connecting = false;
}
/**
* 客户端连接
*/
initClient(config) {
const { url, groupId, key, password, topic: { publish: publishTopic }} = config;
return new Promise((resolve) => {
this.client = mqtt.connect(
{
url,
clientId: `${groupId}@@@${deviceId}`,
username: key,
password,
}
);
this.client.on('connect', () => {
this.connecting = true;
resolve(this);
});
});
}
/**
* 订阅topic
*/
subscribeTopic(topic, config) {
if (this.connecting) {
this.client.subscribe(topic, config);
}
return this;
}
/**
* 发送消息
*/
publishMessage(message) {
this.client.publish(publishTopic, message, { qos: 1 });
}
/**
* 接收消息
*/
handleMessage(callback) {
if (!this.client._events.message) {
this.client.on('message', callback);
}
}
} |
客户端发包函数sendPacketmqtt-packet生成一个可传输buffervar mqtt = require('mqtt-packet')
var object = {
cmd: 'publish',
retain: false,
qos: 0,
dup: false,
length: 10,
topic: 'test',
payload: 'test' // Can also be a Buffer
}
var opts = { protocolVersion: 4 } // default is 4. Usually, opts is a connect packet
console.log(mqtt.generate(object))
// Prints:
//
// <Buffer 30 0a 00 04 74 65 73 74 74 65 73 74>
//
// Which is the same as:
//
// new Buffer([
// 48, 10, // Header (publish)
// 0, 4, // Topic length
// 116, 101, 115, 116, // Topic (test)
// 116, 101, 115, 116 // Payload (test)
// ])sendPacket函数发出packetsend事件并且通过mqtt.writeToStream将packet写入client的stream中。 var mqttPacket = require('mqtt-packet')
function sendPacket (client, packet) {
client.emit('packetsend', packet)
mqttPacket.writeToStream(packet, client.stream, client.options)
}_sendPack方法MqttClient.prototype._sendPacket = function (packet) {
sendPacket(this, packet);
} |
客户端连接 mqtt.connect()
mqtt.connect([url], options)官方说明：
再来看一下我手上项目的连接配置，连接结果。 连接配置 {
key: 'xxxxxxxx',
secret: 'xxxxxxxx',
url: 'wss://foo-bar.mqtt.baz.com/mqtt',
groupId: 'FOO_BAR_BAZ_GID',
topic: {
publish: 'PUBLISH_TOPIC',
subscribe: ['SUBSCRIBE_TOPIC/noticePC/', 'SUBSCRIBE_TOPIC/p2p'],
unsubscribe: 'SUBSCRIBE_TOPIC/noticeMobile/',
},
}
连接结果包括总览，响应头和请求头。 GeneralResponse HeaderRequest Header源码分析下面来看这段mqtt连接的代码。 this.client = mqtt.connect(
{
url,
clientId: `${groupId}@@@${deviceId}`,
username: key,
password,
}
);function parseAuthOptions (opts) {
var matches
if (opts.auth) {
matches = opts.auth.match(/^(.+):(.+)$/)
if (matches) {
opts.username = matches[1]
opts.password = matches[2]
} else {
opts.username = opts.auth
}
}
}
/**
* connect - connect to an MQTT broker.
*
* @param {String} [brokerUrl] - url of the broker, optional
* @param {Object} opts - see MqttClient#constructor
*/
function connect (brokerUrl, opts) {
if ((typeof brokerUrl === 'object') && !opts) {
// 可以传入一个单对象，既包含url又包含选项
opts = brokerUrl
brokerUrl = null
}
opts = opts || {}
// 设置username和password
parseAuthOptions(opts)
if (opts.query && typeof opts.query.clientId === 'string') {
// 设置Client Id
opts.clientId = opts.query.clientId
}
function wrapper (client) {
...
return protocols[opts.protocol](client, opts)
}
// 最终返回一个mqtt client实例
return new MqttClient(wrapper, opts)
} |
订阅topic mqtt.Client#subscribe()实际代码const topic = {
subscribe: ['SUBSCRIBE_TOPIC/noticePC/', 'SUBSCRIBE_TOPIC/p2p'],
unsubscribe: 'SUBSCRIBE_TOPIC/noticeMobile/',
};
const config = { qos:1 };
this.client.subscribe(topic.subscribe, config)源码分析MqttClient.prototype.subscribe = function () {
var packet
var args = new Array(arguments.length)
for (var i = 0; i < arguments.length; i++) {
args[i] = arguments[i]
}
var subs = []
// obj为订阅的topic列表
var obj = args.shift()
// qos等配置
var opts = args.pop()
var defaultOpts = {
qos: 0
}
opts = xtend(defaultOpts, opts)
// 数组类型的订阅的topic列表
if (Array.isArray(obj)) {
obj.forEach(function (topic) {
if (!that._resubscribeTopics.hasOwnProperty(topic) ||
that._resubscribeTopics[topic].qos < opts.qos ||
resubscribe) {
var currentOpts = {
topic: topic,
qos: opts.qos
}
// subs是最终的订阅的topic列表
subs.push(currentOpts)
}
})
}
// 这个packet很重要
packet = {
// 发出订阅命令
cmd: 'subscribe',
subscriptions: subs,
qos: 1,
retain: false,
dup: false,
messageId: this._nextId()
}
// 发出订阅包
this._sendPacket(packet)
return this
} |
发送消息 mqtt.Client#publish()实际代码const topic = {
publish: 'PUBLISH_TOPIC',
};
const messge = {
foo: '',
bar: '',
baz: '',
...
}
const msgStr = JSON.stringify(message);
this.client.publish(topic.publish, msgStr);注意publish的消息需要使用JSON.stringify进行序列化，然后再发到指定的topic。 源码分析MqttClient.prototype.publish = function (topic, message, opts, callback) {
var packet
var options = this.options
var defaultOpts = {qos: 0, retain: false, dup: false}
opts = xtend(defaultOpts, opts)
// 将消息传入packet的payload
packet = {
cmd: 'publish',
topic: topic,
payload: message,
qos: opts.qos,
retain: opts.retain,
messageId: this._nextId(),
dup: opts.dup
}
// 处理不同qos
switch (opts.qos) {
case 1:
case 2:
// 发出publish packet
this._sendPacketI(packet);
...
default:
this._sendPacket(packet);
...
}
return this
} |
接收消息 mqtt.Client “message”事件实际代码this.client.on('message', callback);数据以callback的方式接收。 function (topic, message, packet) {}topic代表接收到的topic，buffer则是具体的数据。 handleMessage(callback) {
this.client.on('message', callback);
}
this.client.handleMessage((topic, buffer) => {
let receiveMsg = null;
try {
receiveMsg = JSON.parse(buffer.toString());
} catch (e) {
receiveMsg = null;
}
if (!receiveMsg) {
return;
}
...do something with receiveMsg...
});源码分析MqttClient使用inherits包继承了EventEmitter。 inherits(MqttClient, EventEmitter)那么到底是在哪里间发出message事件的呢？>emit the message event
1.基于websocket-stream建立websocket连接this.stream = this.streamBuilder(this)
function streamBuilder (client, opts) {
return createWebSocket(client, opts)
}
var websocket = require('websocket-stream')
function createWebSocket (client, opts) {
var websocketSubProtocol =
(opts.protocolId === 'MQIsdp') && (opts.protocolVersion === 3)
? 'mqttv3.1'
: 'mqtt'
setDefaultOpts(opts)
var url = buildUrl(opts, client)
return websocket(url, [websocketSubProtocol], opts.wsOptions)
}2. 使用pipe连接基于readable-stream.Writable创建的可写流var Writable = require('readable-stream').Writable
var writable = new Writable()；
this.stream.pipe(writable)；3.nextTick调用_handlePacketwritable._write = function (buf, enc, done) {
completeParse = done
parser.parse(buf)
work()
}
function work () {
var packet = packets.shift()
if (packet) {
that._handlePacket(packet, nextTickWork)
}
}
function nextTickWork () {
if (packets.length) {
process.nextTick(work)
} else {
var done = completeParse
completeParse = null
done()
}
}4. 在handlePacket中调用handlePublish发出message事件MqttClient.prototype._handlePacket = function (packet, done) {
switch (packet.cmd) {
case 'publish':
this._handlePublish(packet, done)
break
...
}
// emit the message event
MqttClient.prototype._handlePublish = function (packet, done) {
switch (qos) {
case 1: {
// emit the message event
if (!code) { that.emit('message', topic, message, packet) }
}
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
先来看下MQTT在物联网领域的应用场景：
mqtt.js是MQTT在nodejs端的实现。
通过npm package.json包管理，现代vue技术栈下的前端也可用，比如用vue-cli，create-react-app等等构建的项目。
mqtt.js官方为微信小程序和支付宝小程序也做了支持。微信小程序的MQTT协议名为
wxs，支付宝小程序则是alis。如果还是一脸懵逼，那么就跟随我通过mqtt.js去认识一下这个物联网领域的宠儿吧。
The text was updated successfully, but these errors were encountered: