基于mqtt的消息推送(二)服务端实现

基于Mosca的服务端实现

Mosca简介,Mosca基于node.js开发,特性引用自项目首页介绍如下:

Features

  • MQTT 3.1 and 3.1.1 compliant.
  • QoS 0 and QoS 1.
  • Various storage options for QoS 1 offline packets, and subscriptions.
  • As fast as it is possible.
  • Usable inside ANY other Node.js app.
  • Supports Node.js v0.10 and v0.12.
  • Supports io.js v1.x and v2.x and v3.x (please do not use v3.1.0)

需要特别说明的是Mosca并不支持Qos 2。

为什么选择Mosca?首先基于node.js开发,上手难度相对较小,其次协议支持完整,除了不支持Qos 2,其它的基本都支持。持久化支持redis以及mongo。二次开发接口简单。部署非常简单,并且支持docker镜像。

开发步骤

安装Mosca

本文环境假设你已经安装了node.js环境以及redis,我本机的环境如下:mac ox 10.11.2 node.js v0.12 redis最新版本。

Mosca项目托管地址:https://github.com/mcollina/mosca
官方wiki:https://github.com/mcollina/mosca/wiki

安装非常简单,源引自官方说明如下:

Standalone

1
2
npm install mosca bunyan -g
mosca -v | bunyan

Embedded

1
npm install mosca --save

启动脚本

官方给了一个简单的例子,如下是我现在使用的测试代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
var mosca = require('mosca')

//ascoltatore是Mosca作者开发的一个订阅与发布类库,Mosca核心的订阅与发布模型
var ascoltatore = {
type: 'redis', //指定类型,mongo的写法请参考官方wiki
redis: require('redis'),
db: 1,
port: 6379,
return_buffers: true, // to handle binary payloads
//指定数据保留多长时间,单位毫秒
ttl: {
// TTL for subscriptions is 23 hour
subscriptions: 23 * 60 * 60 * 1000,
// TTL for packets is 23 hour
packets: 23 * 60 * 60 * 1000,
},
host: "localhost"
};

//基本参数设置
var moscaSettings = {
port: 1883, //设置监听端口
backend: ascoltatore,
maxInflightMessages: 10240, //设置单条消息的最大长度,超出后服务端会返回
//设置WebSocket参数
http: {
port: 1884,
bundle: true,
static: './'
},
//数据持久化参数设置
persistence: {
factory: mosca.persistence.Redis,
db: 1,
port: 6379,
return_buffers: true, // to handle binary payloads
ttl: {
// TTL for subscriptions is 23 hour
subscriptions: 23 * 60 * 60 * 1000,
// TTL for packets is 23 hour
packets: 23 * 60 * 60 * 1000,
},
host: "localhost"
}
};

//如果需要用户登录验证权限,需要改写此方法
//这里以简单判断了用户名和密码为例,真实环境可以连接实际业务系统的鉴权服务
var authenticate = function(client, username, password, callback) {
var authorized = (username === 'test' && password.toString() === 'passwd');
if (authorized) client.user = username;
callback(null, authorized);
}

function authPub(client, topic, payload, callback) {
callback(null, payload);
}

function authSub(client, topic, callback) {
callback(null, topic);
}

var server = new mosca.Server(moscaSettings);
server.on('ready', setup);

server.on('clientConnected', function(client) {
console.log('client connected', client.id);
});

server.on('published', function(packet, client) {
console.log('Published', packet.topic + packet.payload);
});

// fired when the mqtt server is ready
function setup() {
server.authenticate = authenticate;
server.authorizePublish = authPub;
server.authorizeSubscribe = authSub;

console.log('Mosca server is up and running')
}

二次开发

可以监听的事件列表

  • clientConnected : when a client is connected; the client is passed as a
    parameter.
  • clientDisconnecting : when a client is being disconnected; the client is
    passed as a parameter.
  • clientDisconnected : when a client is disconnected; the client is passed as
    a parameter.
  • published : when a new message is published; the packet and the client are
    passed as parameters.
  • delivered : when a client has sent back a puback for a published message; the packet and the client are
    passed as parameters.
  • subscribed : when a client is subscribed to a topic; the topic and the
    client are passed as parameters.
  • unsubscribed : when a client is unsubscribed to a topic; the topic and the
    client are passed as parameters.

有了上面可以监听到事件你就可以根据自己的业务进行相应的开发,拦截特定的事件并添加业务代码

ascoltatore托管地址 https://github.com/mcollina/ascoltatori

高级参数设置可以参考 https://github.com/mcollina/mosca/wiki/Mosca-advanced-usage

权限验证可以参考 https://github.com/mcollina/mosca/wiki/Authentication-&-Authorization

配置ssl可以参考 https://github.com/mcollina/mosca/wiki/TLS-SSL-Configuration

配置WebSocket可以参考 https://github.com/mcollina/mosca/wiki/MQTT-over-Websockets

写在最后

感谢Mosca的作者mcollina,Mosca非常强大,并且足够简单。下篇文章会介绍如何利用IOS和Android开源客户端类库与Mosca对接,敬请期待!