mosquitto 与 websocket 的结合

摘要：前言作为一个消息代理客户端与服务端的通信时基于协议的而现在的主流应用时呈现在浏览器中这意味着用户与服务端只能通过或者这类浏览器能理解的协议传输所以后端还要建立一个代理层将协议传输的内容解析一下以协议发送到最后再由发送到硬件端在浏览器支持的协

mosquitto 作为一个消息代理, 客户端与 mosquitto 服务端的通信时基于 MQTT 协议的, 而现在的主流 web 应用时呈现在浏览器中, 这意味着用户与服务端只能通过 HTTP 或者 HTTPS 这类浏览器能理解的协议传输, 所以后端还要建立一个代理层, 将 HTTP 协议传输的内容解析一下以 MQTT 协议发送到 mosquitto, 最后再由 mosquitto 发送到硬件端.

在浏览器支持的协议中, 还有一个适用于长连接的 WS 协议, 参考: 浏览器中常见网络协议介绍. 如果客户端直接通过 websocket 连接到 mosquitto 端, 那么就不需要中间的后端代理层. 后端只需要一个推送服务和控制系统就可以实现对客户端的监听, 控制, 推送(当然, 如果业务量巨大, 业务逻辑复杂, 代理层还是有必要的, 因为这样不但可以为 mosquitto 过滤一些不必要的业务, 而且可以做一些数据统计, 设计事件钩子).

在编译 mosquitto 和它的验证插件 mosquitto-auth-plug 的时候, 我注意到 mosquitto 有一个监听的 9001 端口, 事后查了一下, 发现 mosquitto 开放的这个端口是支持直接与客户端进行 websocket 通信的.

其实纯粹的 MQTT 服务器是没有这个功能的, mosquitto 需要在编译的时候设置 configure.mk中

WITH_WEBSOCKETS := yes

所幸的是, eclipse 官方的 docker 镜像 已经支持了 websocket 通信方式. 只需要在 mosquitto.conf 中启用:

port 1883
listener 9001
protocol websockets

eclipse 提供了用于 浏览器客户端利用 javascript 和 mosquitto 进行 websocket 通信的 paho-mqtt.js, 这是这个 js 库的文档: Paho.MQTT DOC.

我们假设有客户端用户为 client, 服务端用户为 server. 为他们分配的主题与权限为:

 id | username |      topic         | rw 
----+----------+--------------------+----
  1 | client   | /p/client/upload   |  2
  2 | server   | /p/client/upload   |  2
  3 | client   | /p/client/download |  1
  4 | server   | /p/client/download |  2

那么 server 和 client 均可以在 /p/client/upload 读写. 我们约定只有 client 在 /p/client/upload 写, server 只读 /p/client/upload, 这个 topic 是用于 client 向 server 发送消息.

对于 topic /p/client/download, server 可读可写, client 只读, 这个 topic 是用于 client 接受 server 向下推送的消息, 具体的逻辑如下:

注意这里的 client 是对 /p/client/upload 具有可读可写权限的, 意味者它可以读取来自 /p/client/upload 的信息, 但是实际上这里没有人往这个主题发布消息(除非 client 自己往这里发消息), 所以 client 拥有可读权限也没有关系.

建议看一下这篇文章: Using MQTT Over WebSockets with Mosquitto, 它比较详细地介绍了客户端通过 websocket 于 mosquitto 服务器通信的方式.
比如对于 client.html:
先引入 paho-mqtt.js, 这里我们先用 cloudflare.com CDN上的这段 js:

下面是创建客户端 websocket 的例子:

var mqtt;
var host = "mosquitto";
var port = 9001;

// onConnect 事件
function onConnect() {
    console.log("connected.");
    var raw_message = "Hello World!";
    message = new Paho.MQTT.Message(raw_message);
    message.destinationName = "/p/client/upload";
    console.log("sending message: " + raw_message );
    mqtt.send(message);

    // 订阅 download topic
    var subOptions = {
        qos: 1,
        onSuccess: onSubscribe
    };
    mqtt.subscribe("/p/client/download", subOptions);
}

// 订阅主题成功事件
function onSubscribe(context) {
    console.log("subscribe success");
    console.log(context);
}

// 连接失败事件
function onFailure(message) {
    console.log("connect failed.");
}

// onMessageArrived 事件
function onMessageArrived(message) {
    console.log("new message arrived...");
    console.log(message.payloadString);
}


// 建立 MQTT websocket 连接
function MQTTconnect() {
    console.log("connecting to " + host + ":" + port);
    mqtt = new Paho.MQTT.Client(host, port, "clientid");
    var options = {
        timeout: 3,
        onSuccess: onConnect,
        onFailure: onFailure,
        userName: "client",
        password: "123456",
        mqttVersion: 4
    };
    mqtt.onMessageArrived = onMessageArrived;
    mqtt.connect(options);
}

这里我们利用 paho-mqtt.js 新建了一个 mqtt, 然后绑定 onSuccess(连接成功), onFailure(连接失败) onMessageArrived(消息到来)等事件, 之后用 options 里的配置连接到远程的 mosquitto 服务器. 连接成功后, client 向 server 发送一条消息到 /p/client/upload topic, 通知服务端已经建立连接. 发送之后 mqtt 又订阅 /p/client/download, 准备接受来自服务端的信息.

比如, 服务端可以这样向客户端推送消息:

import paho.mqtt.publish as publish
import time

HOST = "mosquitto"
PORT = 1883


if __name__ == "__main__":
    client_id = time.strftime("%Y%m%d%H%M%S", time.localtime(time.time()))
    publish.single(
        "/p/client/download", "hello mqtt", qos=2, hostname=HOST, port=PORT,
        client_id=client_id, auth={"username": "server", "password": "123456"})

这就是基于双通道的服务端于客户端通信

单通道通信原理类似, 但是由于客户端可能有多台设备, 比如手机端, 微信小程序端, PC 端. 那么身份的验证不能直接由 mosquitto 确定, 应该在消息体内确定. 比如我们用 json 作为消息的承载方式. 消息体可以这样:

{
    "timestamp": "1539141568",
    "client_id": "1001001",
    "message_type": "ping",
    "data": {"detail": "content"},
    "token": "auth_token"
}

这里应该按照具体的业务需求进行鉴权设计.