思享工具箱

EMQ X 插件持久化系列 （四）PostgreSQL 数据持久化

摘要：
作为开源关系数据库的重要成员，PostgreSQL号称它是世界上最先进的开源数据库。与其他开源关系数据库（如MySQL）相比，PostgreSQL是一个完全由社区驱动的开源项目，由全球1000多名贡献者维护。PostgreSQL还具有强大的SQL编程功能，如统计函数和统计语法支持。通过Timescaledb插件，PostgreSQL可以转换为功能齐全的时序数据库Timescaledb。存储客户端联机状态时，插件会将联机状态、联机/脱机时间和节点客户端列表更新到PostgreSQL数据库。

本章节以在 CentOS 7.2 中的实际例子来说明如何通过 PostgreSQL 来存储相关的信息。

作为开源关系数据库重要一员，PostgreSQL 标榜自己是世界上最先进的开源数据库，相比于其他开源关系数据库如 MySQL，PostgreSQL 是完全由社区驱动的开源项目，由全世界超过 1000 名贡献者所维护。PostgreSQL 提供了单个完整功能的版本，而不像 MySQL 那样提供了多个不同的社区版、商业版与企业版。PostgreSQL 基于自由的 BSD/MIT 许可，组织可以使用、复制、修改和重新分发代码，只需要提供一个版权声明即可。

PostgreSQL 具有诸多特性，在 GIS 领域有较多支持，其“无锁定”特性非常突出，支持函数和条件索引，有成熟的集群方案。PostgreSQL 还具备及其强悍的 SQL 编程能力如统计函数和统计语法支持，通过 Timescaledb 插件，PostgreSQL 可以转变为功能完备的时序数据库 Timescaledb 。

功能概览

  • 客户端在线状态存储
  • 客户端代理订阅
  • 持久化发布消息
  • Retain 消息持久化
  • 消息确认持久化
  • 自定义 SQL

安装与验证 PostgreSQL 服务器

读者可以参考 PostgreSQL 官方文档 或 Docker 来下载安装PostgreSQL 服务器，本文章使用 PostgreSQL 10.1 版本。

为方便管理操作，可下载使用免费图形化管理软件 Postico（仅限 MacOS）或 pgAdmin。

配置 EMQ X 服务器

通过 RPM 方式安装的 EMQ X，PostgreSQL 相关的配置文件位于 /etc/emqx/plugins/emqx_backend_pgsql.conf，如果只是测试 PostgreSQL 持久化的功能，大部分配置不需要做更改，填入用户名、密码、数据库即可：

backend.pgsql.pool1.server = 127.0.0.1:5432

backend.pgsql.pool1.pool_size = 8

backend.pgsql.pool1.username = root

backend.pgsql.pool1.password = public

backend.pgsql.pool1.database = mqtt

backend.pgsql.pool1.ssl = false

保持剩下部分的配置文件不变，然后需要启动该插件。启动插件的方式有 命令行和 控制台两种方式，读者可以任选其一。

通过命令行启动

emqx_ctl plugins load emqx_backend_pgsql

通过管理控制台启动

EMQ X 管理控制台 插件 页面中，找到 emqx_backend_pgsql 插件，点击 启动。

客户端在线状态存储

客户端上下线时，插件将更新在线状态、上下线时间、节点客户端列表至 PostgreSQL 数据库。

数据表

创建 mqtt_client 设备在线状态表:

CREATE TABLE mqtt_client(
  id SERIAL primary key,
  clientid character varying(100),
  state integer, -- 在线状态: 0 离线 1 在线
  node character varying(100), -- 接入节点名称
  online_at timestamp, -- 上线时间
  offline_at timestamp, -- 下线时间
  created timestamp without time zone,
  UNIQUE (clientid)
);

配置项

打开配置文件，配置 Backend 规则：

## hook: client.connected、client.disconnected
## action/function: on_client_connected、on_client_disconnected


## 客户端上下线
backend.pgsql.hook.client.connected.1 = {"action": {"function": "on_client_connected"}, "pool": "pool1"}

## 客户端下线
backend.pgsql.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}

使用示例

浏览器打开 http://127.0.0.1:18083 EMQ X 管理控制台，在 工具 -> Websocket 中新建一个客户端连接，指定 clientid 为 sub_client，点击连接，连接成功后手动断开:

image20181116105333637.png

查看 mqtt_client 表，此时将写入 / 更新一条客户端上下线记录：

1.png

客户端代理订阅

客户端上线时，存储模块直接从数据库读取预设待订阅列表，代理加载订阅主题。在客户端需要通过预定主题通信（接收消息）场景下，应用能从数据层面设定 / 改变代理订阅列表。

数据表

创建 mqtt_sub 代理订阅关系表:

CREATE TABLE mqtt_sub(
  id SERIAL primary key,
  clientid character varying(100),
  topic character varying(200), -- topic
  qos integer, -- QoS
  created timestamp without time zone,
  UNIQUE (clientid, topic)
);

配置项

打开配置文件，配置 Backend 规则：

## hook: client.connected
## action/function: on_subscribe_lookup
backend.pgsql.hook.client.connected.2    = {"action": {"function": "on_subscribe_lookup"}, "pool": "pool1"}

使用示例

当 sub_client 设备上线时，需要为其订阅 sub_client/upstream 与 sub_client/downlink 两个 QoS 1 的主题：

  1. 在 mqtt_sub 表中初始化插入代理订阅主题信息：
insert into mqtt_sub(clientid, topic, qos) values('sub_client', 'sub_client/upstream', 1);

insert into mqtt_sub(clientid, topic, qos) values('sub_client', 'sub_client/downlink', 1);
  1. EMQ X 管理控制台 WebSocket 页面，以 clientid sub_client 新建一个客户端连接，切换至订阅页面，可见当前客户端自动订阅了 sub_client/upstream 与 sub_client/downlink 两个 QoS 1 的主题：

image20181116110036523.png

  1. 切换回管理控制台 WebSocket 页面，向 sub_client/downlink 主题发布消息，可在消息订阅列表收到发布的消息。

持久化发布消息

数据表

创建 mqtt_msg MQTT 消息持久化表:

CREATE TABLE mqtt_msg (
  id SERIAL primary key,
  msgid character varying(60),
  sender character varying(100), -- 消息 pub 的 clientid
  topic character varying(200),
  qos integer,
  retain integer, -- 是否 retain 消息
  payload text,
  arrived timestamp without time zone -- 消息抵达时间(QoS > 0)
);

配置项

打开配置文件，配置 Backend 规则，支持使用 topic 参数进行消息过滤，此处使用 # 通配符存储任意主题消息：

## hook: message.publish
## action/function: on_message_publish

backend.pgsql.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}

使用示例

在 EMQ X 管理控制台 WebSocket 页面中，使用 clientdi sub_client 建立连接，向主题 upstream_topic 发布多条消息，EMQ X 将消息列表持久化至 mqtt_msg 表中：

websocket.png

暂只支持 QoS 1 2 的消息持久化。

Retain 消息持久化

表结构

创建 mqtt_retain Retain 消息存储表:

CREATE TABLE mqtt_retain(
  id SERIAL primary key,
  topic character varying(200),
  msgid character varying(60),
  sender character varying(100),
  qos integer,
  payload text,
  arrived timestamp without time zone,
  UNIQUE (topic)
);

配置项

打开配置文件，配置 Backend 规则：

## 同时开启以下规则，启用 retain 持久化三个生命周期

## 发布非空 retain 消息时 (存储)
backend.pgsql.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}

## 设备订阅主题时查询 retain 消息
backend.pgsql.hook.session.subscribed.2  = {"topic": "#", "action": {"function": "on_retain_lookup"}, "pool": "pool1"}

## 发布空 retain 消息时 (清除)
backend.pgsql.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}

使用示例

在 EMQ X 管理控制台 WebSocket 页面中建立连接后，发布消息勾选保留：

image20181119111926675.png

发布（消息不为空）

非空的 retain 消息发布时，EMQ X 将以 topic 为唯一键，持久化该条消息至 mqtt_retain 表中，相同主题下发布不同的 retain 消息，只有最后一条消息会被持久化：

image20181119112306703.png

订阅

客户端订阅 retain 主题后，EMQ X 将查询 mqtt_retain 数据表，执行投递 retain 消息操作。

发布（消息为空）

MQTT 协议中，发布空的 retain 消息将清空 retain 记录，此时 retain 记录将从 mqtt_retain 表中删除。

消息确认持久化

开启消息确认 (ACK) 持久化后，客户端订阅 QoS 1、QoS 2 级别的主题时，EMQ X 将在数据库以 clientid + topic 为唯一键初始化 ACK 记录。

数据表

创建 mqtt_acked 消息确认表:

CREATE TABLE mqtt_acked (
  id SERIAL primary key,
  clientid character varying(100),
  topic character varying(100),
  mid integer,
  created timestamp without time zone,
  UNIQUE (clientid, topic)
);

配置项

打开配置文件，配置 Backend 规则，可使用 topic 通配符 过滤要应用的消息：

## 订阅时初始化 ACK 记录
backend.pgsql.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "pool": "pool1"}


## 消息抵达时更新抵达状态
backend.pgsql.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked"}, "pool": "pool1"}

## 取消订阅时删除记录行
backend.pgsql.hook.session.unsubscribed.1= {"topic": "#", "action": {"sql": ["delete from mqtt_acked where clientid = ${clientid} and topic = ${topic}"]}, "pool": "pool1"}

使用示例

在 EMQ X 管理控制台 WebSocket 页面中建立连接后，订阅 QoS > 0 的主题：

image20181119140251843.png

此时 mqtt_acked 表将插入初始化数据行，每向主题发布一条 QoS > 0 的消息，消息抵达后数据行 mid 将自增 1：

image20181119165248998.png

代理订阅中满足 QoS > 0 的 topic 也会初始化记录，客户端取消订阅后相关记录将被删除。

自定义 SQL

除去插件内置函数、表结构外，emqx_backend_pgsql 还支持自定义 SQL 语句，通过使用如 ${clientid} 模板语法动态构造 SQL 语句实现如客户端连接历史、更新自定义数据表等操作。

SQL语句参数说明

hook可用参数示例(sql语句中${name} 表示可获取的参数)
client.connectedclientidinsert into conn(clientid) values(${clientid})
client.disconnectedclientidinsert into disconn(clientid) values(${clientid})
session.subscribedclientid, topic, qosinsert into sub(topic, qos) values(${topic}, ${qos})
session.unsubscribedclientid, topicdelete from sub where topic = ${topic}
message.publishmsgid, topic, payload, qos, clientidinsert into msg(msgid, topic) values(${msgid}, ${topic})
message.ackedmsgid, topic, clientidinsert into ack(msgid, topic) values(${msgid}, ${topic})
message.deliveredmsgid, topic, clientidinsert into delivered(msgid, topic) values(${msgid}, ${topic})

更新自定义数据表示例

应用现有设备表 clients，具有设备连接认证、设备状态记录、设备管理等基本字段用于其他管理业务，现需要将 EMQ X 设备状态同步至该表中：

CREATE TABLE "public"."clients" (
    "id" serial,
    "deviceUsername" varchar(50), --  MQTT username
    "client_id" varchar(50), -- MQTT client_id
    "password" varchar(50), -- MQTT password
    "is_super" boolean DEFAULT 'false', -- 是否 ACL super 客户端
    "owner" int, -- 创建用户
    "productID" int, -- 所属产品
    "state" boolean DEFAULT 'false', -- 在线状态
    PRIMARY KEY ("id")
);

-- 初始化系统中已存在示例数据，此时 state 为 false
INSERT INTO "public"."clients"("deviceUsername", "client_id", "password", "is_super", "owner", "productID", "state") VALUES('mqtt_10c61f1a1f47', 'mqtt_10c61f1a1f47', '9336EBF25087D91C818EE6E9EC29F8C1', TRUE, 1, 21, FALSE);

自定义 UPDATE SQL 语句：

## connected / disconnected hook 中配置自定义 UPDATE SQL
## 可以配置多条 SQL 语句 "SQL": ["sql_a", "sql_b", "sql_c"]

## 连接时
backend.pgsql.hook.client.connected.3 = {"action": {"sql": ["update clients set state = true where client_id = ${clientid}"]}, "pool": "pool1"}

## 断开时
backend.pgsql.hook.client.disconnected.3 = {"action": {"sql": ["update clients set state = false where client_id = ${clientid}"]}, "pool": "pool1"}

客户端上线时将填充并执行预定的 SQL 语句，更新设备在线状态 state 字段为 true：

image20181119170648517.png

高级选项

backend.pgsql.time_range = 5s

backend.pgsql.max_returned_count = 500

总结

读者在理解了 PostgreSQL 中所存储的数据结构、自定义 SQL 之后，可以结合 PostgreSQL 拓展相关应用。

标签：#topic#postgresql#数据持久化#mqtt协议#qos#sub | 浏览：287 | 发布日期：

免责声明：文章转载自《EMQ X 插件持久化系列 （四）PostgreSQL 数据持久化》仅用于学习参考。如对内容有疑问，请及时联系本站处理。

上篇Ubuntu环境下使用Unity3D调用C++编写的Linux so动态链接库android studio连接雷电模拟器下篇

宿迁高防，2C2G15M，22元/月；香港BGP，2C5G5M，25元/月 雨云优惠码：MjYwNzM=

相关文章

K8s--09 编写mysql的持久化deployment

k8s实战--编写mysql的持久化deployment 1.实验-编写mysql的持久化deployment #先打标签 [root@node1 ~/volume]# kubectl label nodes node3 disktype=SSD #编写资源清单 [root@node1 ~/volume]# cat mysql-hostpath-ssd....

NSQ部署

一、 简介 NSQ主要有三个主要程序和一个Web服务程序： nsqd：是守护进程，接收，缓存，并投递消息给客户端 nsqlookupd：是一个守护进程，为消费者提供运行时发现服务，来查找指定话题（topic）的生产者 nsqd nsq_to_http：消费指定的话题（topic）/通道（channel）和执行 HTTP requests (GET/POS...

MQTT 单片机端讲解

 　　　　有空了和大家分享一下，如何从头架构一个高效mqtt并行客户端，基于传统GPRS等较差网络环境和网关等网络环境好的情景（当然仔细讲解mqtt的基本函数使很有必要的）。---这会正忙着搬砖 MQTt协议 第一个字节用来表示报表类型   接下来第二个字节开始是剩余长度 表示长度，最多用4个字节，变长，每个字节的高位表示后面是否还有表示长度的字节...

使用VBA数组公式——Excel之VBA(11)

一、 认识VBA数组及常用操作 引例：计算所属区域的总金额 Sub test() Dim i, k Dim t t = Timer '获取时间值 For i = 2 To 200000 If Range("g" & i) = Range("n5") Then k = k + Range("j" & i)...

PostgreSQL表空间_数据库_模式_表_用户角色之间的关系[转]

看PostgreSQL9的官方文档，我越看越迷糊，这表空间，数据库，模式，表，用户，角色之间的关系怎么在PostgreSQL里这么混乱呢？ 经过中午的一个小实验，我逐渐理清了个中来龙去脉。下面我来还原我的试验，并循序讲清其中关系。 首先，实验出角色与用户的关系 在PostgreSQL中，存在两个容易混淆的概念：角色/用户。之所以说这两个概念容易混淆，是因...

len()函数

简单的说，就是读取一个字符串的 字符长度 的函数。 实例1： 1 Sub W1() 2 If Len(Dir("d:/A.xls")) = 0 Then 3 MsgBox "A文件不存在" 4 Else 5 MsgBox "A文件存在" 6 End If 7 End Sub 实例2： 1 For x...

最新文章

随机推荐

思享工具箱导航