写在前面:
前两篇文字<<基于MQTT协议谈谈物联网开发-华佗写代码>>,<<基于MQTT协议实现Broker-华佗写代码>>主要叙述了MQTT协议的编解码以及基于MQTT协议的一些常见应用场景,并以一个简单的消息推送系统作为例子具体阐述了Mqtt Broker部分的实现,之前主要以原生android或者iOS或者服务端代理作为例子,考虑到在移动端开发时,选择的技术栈有所不同,有的选择web前端开发.作为例子,这里以之前的消息推送系统为例基于web前端开发,继续叙述基于WebSocket协议实现Broker.
1.WebSocket协议主要特点:
(1)基于http协议握手建立tcp长连接;
(2)相比http,WebSocket协议交换最小化,降低网络流量;
(3)双向通信,服务器可以主动推送数据给客户端;
2.Mqtt Broker具体实现(WebSocket部分):
2.1Mqtt Broker架构草图:
2.2Mqtt Broker实现细节:
(1)新增实现websocket server,监听不同端口;
(2)每个websocket连接,实例化一个mqttclient负责其协议解析,消息发布和订阅等;
(3)复用之前Mqtt Broker与RabbitMQ通信部分,具体参考上一篇文字;
(4)其他...
2.3Mqtt Broker代码实现(WebSocket部分):
type tcpKeepAliveListener struct {
*net.TCPListener
}
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
}
//监听websocket server地址,注册websocket handler
func (mb *MqttBroker) ListenAndServeWeb() {
defer mb.wg.Done()
http.HandleFunc("/", mb.webHandler)
webserver := &http.Server{Addr: mb.webaddr, Handler: nil}
var listener net.Listener
var err error
listener, err = net.Listen("tcp", mb.webaddr)
if err != nil {
return
}
U.GetLog().Printf("listen and serve web broker on %s", mb.webaddr)
err = webserver.Serve(tcpKeepAliveListener{listener.(*net.TCPListener)})
}
//每一个Websocket连接,实例化一个MqttClient负责其协议解析,以及与rabbitmq的通信
func (mb *MqttBroker) webHandler(w http.ResponseWriter, r *http.Request) {
upgrader.CheckOrigin = checkSameOrigin
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
U.GetLog().Printf("upgrade error:%v", err)
return
}
mqttclient, err := NewMqttClient(mb.wg, mb, nil, conn, "web")
if err != nil {
return
}
mb.clientMap[mqttclient.GetClientID()] = mqttclient
mb.wg.Add(1)
go mqttclient.ServeWeb()
}
2.4Mqtt Client代码实现(WebSocket部分):
//定义WebSocket通信消息格式
//Action选项有publish,subscribe,unsubscribe
type WebMessage struct {
Action string
Topic string
Payload string
}
type MqttClient struct {
wg *sync.WaitGroup
broker *MqttBroker
tcpconn net.Conn
wconn *websocket.Conn
...
needDisConn bool
}
//通过WebMessage.Action区分消息指令类型
func (mc *MqttClient) ServeWeb() {
defer mc.wg.Done()
defer mc.commonDefer()
if mc.wconn == nil {
return
}
for {
if mc.needDisConn {
break
}
_, message, err := mc.wconn.ReadMessage()
if err != nil {
U.GetLog().Printf("handle message error:%v", err)
mc.needDisConn = true
continue
}
wm := WebMessage{}
err = json.Unmarshal(message, &wm)
if err != nil {
U.GetLog().Printf("json.Unmarshal(message, &wm) error:%v", err)
continue
}
switch wm.Action {
case "subscribe":
err = mc.handleWebSubscibe(wm.Topic)
case "publish":
err = mc.handleWebPublish(wm.Topic, wm.Payload)
case "unsubscribe":
err = mc.handleWebUnSubscribe(wm.Topic)
case "ping":
mc.lastheartbeat = 0
default:
U.GetLog().Printf("unexpected WebMessage Action:%s", wm.Action)
continue
}
if err != nil {
U.GetLog().Printf("handle message error:%v", err)
}
mc.lastheartbeat = 0
}
}
3.WebSocket Client端实现:
3.1实现细节:
(1)建立与WebSocket Server的连接;
(2)初始化WebSocket,注册相关回调函数;
(3)实现WebSocket断线重连机制;
(4)封装类似mqtt基于topic的发布订阅等接口;
(5)Nodejs端需要browserify相关js文件,Javascript端可以直接调用WebSocket;
(6)其他...
3.2具体代码实现:
var WebSocket = require('ws');
var WEBSOCKET_MQTT_BROKER = 'ws://your_server_ip/ws/';
var ping = {
Action: "ping"
};
var _listeners = {};
var _websocket = null;
var _connected = false;
_access = function () {
console.log('try mqtt.connect');
_connect_websocket();
setInterval(function () {
_reconnect_websocket();
if (_websocket != null && _connected) {
_websocket.send(JSON.stringify(ping));
}
}, 3000);
};
//websocket初始化,并实现相关回调函数
_init_websocket = function () {
if (_websocket == null) {
return;
}
_websocket.onopen = function () {
_connected = true;
console.log("Connected to WebSocket server.");
for (var topic in _listeners) {
var sub = {
Action: "subscribe",
Topic: topic,
Payload: ""
};
_websocket.send(JSON.stringify(sub));
}
};
_websocket.onclose = function () {
_connected = false;
_websocket = null;
console.log("Disconnected");
};
_websocket.onmessage = function (evt) {
console.log('recv data from server: ' + evt.data);
var dataObj = JSON.parse(evt.data);
_listeners[dataObj.Topic] && _listeners[dataObj.Topic](dataObj.Payload);
};
_websocket.onerror = function (evt) {
_connected = false;
_websocket = null;
console.log('Error occured: ' + evt);
};
};
_connect_websocket = function () {
if (_connected) {
return;
}
_websocket = new WebSocket(WEBSOCKET_MQTT_BROKER);
_init_websocket();
};
//断线重连,通过定时器实现每三秒断线重连
_reconnect_websocket = function () {
if (_connected) {
return;
}
_websocket = new WebSocket(WEBSOCKET_MQTT_BROKER);
_init_websocket();
};
//模拟mqtt发布消息
sendMessage = function (topic, data) {
if (!_websocket || !_connected) {
var err = new Error('iot client not ready.');
console.warn(err);
return;
}
var send_data = JSON.stringify(data);
var pub = {
Action: "publish",
Topic: topic,
Payload: send_data
};
_websocket.send(JSON.stringify(pub));
};
//模拟mqtt订阅消息,并根据topic注册回调函数
onMessage = function (topic, callback) {
_listeners[topic] = callback;
if (!_websocket || !_connected) {
console.warn('onMessage, but iot client not ready.');
return;
}
var sub = {
Action: "subscribe",
Topic: topic,
Payload: ""
};
_websocket.send(JSON.stringify(sub));
};
//模拟mqtt取消订阅,并根据topic删除对应回调函数
stopReceiveMessage = function (topic) {
delete _listeners[topic];
if (!_websocket || !_connected) {
console.warn('stopReceiveMessage, but iot client not ready.');
return;
}
var unsub = {
Action: "unsubscribe",
Topic: topic,
Payload: ""
};
_websocket.send(JSON.stringify(unsub));
};
_access();
4.WebSocket相关nginx配置:
server {
listen 80;
server_name your_server_name;
...
location /ws/ {
proxy_redirect off;
add_header Access-Control-Allow-Origin *;
add_header Access-Control-Allow-Methods 'GET, POST, OPTIONS';
add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
proxy_pass http://127.0.0.1:2884/;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
}
}
出于篇幅考虑,之前两篇文字叙述过的内容,比如Mqtt Broker其他实现部分以及与RabbitMQ通信部分,都是复用之前的代码逻辑,这里不再赘述,Mqtt Broker中WebSocket部分相当于使用WebSocket协议做了MQTT协议的翻译转换,也有一些成员变量,用到了也不一一具体注释了,主要通过代码关键路径叙述实现的一些细节,如有错误,恳请指出,转载也请注明出处!!!
未完待续...