EMQ X MQTT broker plugin persistence series (4) PostgreSQL data persistence

This chapter uses a practical example from CentOS 7.2 to illustrate how to store related information through PostgreSQL.

As an important member of open source relational databases, PostgreSQL claims to be the most advanced open source database in the world. Compared to other open source relational databases such as MySQL, PostgreSQL is a completely community-driven open source project, maintained by more than 1,000 contributors worldwide . PostgreSQL offers a single, full-featured version, unlike MySQL, which offers multiple different community, business, and enterprise versions. PostgreSQL is based on a free BSD / MIT license, and organizations can use, copy, modify, and redistribute code by providing a copyright notice.

PostgreSQL has a number of features and is well supported in the GIS world. Its "lock-free" feature is very prominent. It supports function and condition indexes, and has a mature clustering solution. PostgreSQL also has powerful SQL programming capabilities such as statistical functions and statistical syntax support. With the Timescaledb plugin, PostgreSQL can be transformed into a fully functional time-series database, Timescaledb.

Functions overview

  • Client online status storage
  • Client Agent Subscription
  • Persist publishing message
  • Retain message persistence
  • Message acknowledgement persistence
  • Custom SQL ## Install and validate PostgreSQL server

Readers can refer to PostgreSQL Official Documentation or Docker to download and install the PostgreSQL server. This article Use PostgreSQL 10.1 version.

To facilitate the management operation, you can download and use the free graphical management software Postico (MacOS only) or pgAdmin.

Configure EMQ X MQTT broker

For EMQ X MQTT broker installed by RPM, the PostgreSQL related configuration file is located in /etc/emqx/plugins /emqx_backend_pgsql.conf. If you only test the persistence function of PostgreSQL, most of the configuration does not need to be changed. You only need to fill in the user name, password and database:

backend.pgsql.pool1.server =

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

Leave the rest of the configuration file unchanged, and then you need to start the plugin. There are two ways to start the plugin: command line and console. The reader can choose one of them.

Start from the command line

emqx_ctl plugins load emqx_backend_pgsql

Start through the management console

In the EMQ X management console plugin page, find emqx_backend_pgsql plugin, click start.

Client online status storage

When the client goes online and offline, the plugin will update the online status, online time, and node client list to the PostgreSQL database.

Data sheet

Create the mqtt_client device online status table:

CREATE TABLE mqtt_client(
  id SERIAL primary key,
  clientid character varying(100),
  state integer, -- online status: 0 offline 1 online
  node character varying(100), -- Access node name
  online_at timestamp, -- online time
  offline_at timestamp, -- offline time
  created timestamp without time zone,
  UNIQUE (clientid)

Configuration item

Open the configuration file and configure the Backend rule:

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

## Client is online
backend.pgsql.hook.client.connected.1 = {"action": {"function": "on_client_connected"}, "pool": "pool1"}

## Client is offline
backend.pgsql.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}


Opens EMQ X management console through the browser, create a new client connection in Tools -> Websocket, specify clientid as sub_client, click on connect, and disconnect manually after successful connection:


View the mqtt_client table, at which point a client online and offline record will be written/updated:


Client Agent Subscription

When the client is online, the storage module directly reads the preset to-be-subscribed list from the database, and the agent loads the subscription topic. In the scenario where the client needs to communicate (receive a message) through a preset topic, the application can set / change the agent subscription list from the data level.

Data sheet

Create mqtt_sub agent subscription relational table:

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)

Configuration item

Open the configuration file and configure the Backend rule:

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


When the sub_client device goes online, it needs to subscribe to the two QoS 1 topics of sub-client/upstream and sub_client/downlink :

  1. Initially insert agent subscription topic information in the mqtt_sub table:
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. In the EMQ X management console WebSocket page, create a new client connection with clientid sub_client, switch to subscription page, it can be seen that the current client automatically subscribes to the two QoS 1 topics of sub_client/upstream and sub_client/downlink:


  1. Switch back to the management console WebSocket page and publish the message to the topic sub_client/downlink. You can receive the published message in the message subscription list.

Persist publishing message

Data sheet

Create mqtt_msg MQTT message persistence table:

CREATE TABLE mqtt_msg (
  id SERIAL primary key,
  msgid character varying(60),
  sender character varying(100), -- Message pub's clientid
  topic character varying(200),
  qos integer,
  retain integer, -- whether to retain the message
  payload text,
  arrived timestamp without time zone -- message arrived time(QoS > 0)

Configuration item

Open the configuration file, configure Backend rules, support message filtering using the topic parameter, and use the wildcard # to store arbitrary topic messages:

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

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


In the EMQ X management console WebSocket page, use clientdi sub_client to establish a connection, publish multiple messages to the topic upstream_topic, and EMQ X persists the message list to the mqtt_msg table:


Only QoS 1 2 message persistence is supported for the time being.

Retain Message persistence

Table Structure

Create the mqtt_retain Retain message storage table:

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)

Configuration item

Open the configuration file and configure the Backend rule:

## Enable the following rules at the same time and start retain persistence for three life cycles

## When a non-empty retain message is published (stored)
backend.pgsql.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}

## Query retain message when device subscribes to topic
backend.pgsql.hook.session.subscribed.2  = {"topic": "#", "action": {"function": "on_retain_lookup"}, "pool": "pool1"}

## When an empty retain message is published (cleared)
backend.pgsql.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}


After establishing a connection on the WebSocket page of the EMQ X management console, publish the message and select Reserve:


Publish (message is not empty)

When a non-empty retain message is published, EMQ X will use topic as the unique key to persist the message to the mqtt_retain table. Different retain messages will be published under the same topic. Only the last message will be persisted:



After the client subscribes to the retain topic, EMQ X will query the mqtt_retain data table to perform the post operation of retain message .

Publish (message is empty)

In the MQTT protocol, publishing an empty retain message will clear the retain record. At this time, the retain record will be deleted from the mqtt_retain table.

Message acknowledgemen persistence

When message acknowledgement (ACK) persistence is enabled and a client subscribes to QoS 1 and QoS 2 topics, EMQ X will initialize the ACK record in the database with clientid + topic as the unique key.

Data sheet

Create the mqtt_acked message acknowledgement table:

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)

Configuration item

Open the configuration file and configure the backend rule. Use the topic wildcard to filter the messages to be applied:

## Initialize ACK records when subscribing
backend.pgsql.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "pool": "pool1"}

## Update arrival status when message arrives
backend.pgsql.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked"}, "pool": "pool1"}

## Delete record rows when unsubscribing
backend.pgsql.hook.session.unsubscribed.1= {"topic": "#", "action": {"sql": ["delete from mqtt_acked where clientid = ${clientid} and topic = ${topic}"]}, "pool": "pool1"}


After establishing a connection in the EMQ X Management Console WebSocket page, subscribe to topics with QoS> 0:


At this point, the mqtt_acked table will be inserted with the initialization data row. At each time a message with a QoS> 0 is issued to the topic , the data row mid will increase by 1 when the message arrives:


Topics in the agent subscription that satisfy QoS> 0 will also initialize the records, and the related records will be deleted after the client cancels the subscription.

Custom SQL

In addition to the built-in functions and table structure of the plugin, emqx_backend_pgsql also supports custom SQL statements. By using template syntax such as ${clientid} to dynamically construct SQL statements, it can implement operations such as client connection history and updating custom data tables.

SQL statement parameter description

hook Available parameters Example ($ (name) in the SQL statement indicates the available parameters)
client.connected clientid insert into conn(clientid) values(${clientid})
client.disconnected clientid insert into disconn(clientid) values(${clientid})
session.subscribed clientid, topic, qos insert into sub(topic, qos) values(${topic}, ${qos})
session.unsubscribed clientid, topic delete from sub where topic = ${topic}
message.publish msgid, topic, payload, qos, clientid insert into msg(msgid, topic) values(${msgid}, ${topic})
message.acked msgid, topic, clientid insert into ack(msgid, topic) values(${msgid}, ${topic})
message.delivered msgid, topic, clientid insert into delivered(msgid, topic) values(${msgid}, ${topic})

Example of Updating custom data sheet

The existing device table clients has basic fields such as device connection authentication, device status record, and device management for other management services. Now we need to synchronize the EMQ X device status to this table:

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', -- Whetner it is ACL super client
    "owner" int, -- Create user
    "productID" int, -- Product
    "state" boolean DEFAULT 'false', -- Online status
    PRIMARY KEY ("id")

-- Sample data already exists in the initialization system, at this time state is false
INSERT INTO "public"."clients"("deviceUsername", "client_id", "password", "is_super", "owner", "productID", "state") VALUES('mqtt_10c61f1a1f47', 'mqtt_10c61f1a1f47', '9336EBF25087D91C818EE6E9EC29F8C1', TRUE, 1, 21, FALSE);

Custom UPDATE SQL statement:

##Configure custom UPDATE SQL in connected / disconnected hook
## You can configure multiple SQL statements "SQL": ["sql_a", "sql_b", "sql_c"]

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

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

When the client goes online, it will fill in and execute the preset SQL statement, and and update the state field of the device online status to true:


Advanced options

backend.pgsql.time_range = 5s

backend.pgsql.max_returned_count = 500


After the reader understands the data structures stored in PostgreSQL and custom SQL, they can expand related applications.

The streaming database built for IoT data storage and real-time processing.

Fully managed MQTT 5.0 IoT cloud, start a 30-day free trial.

Related Links

Integrate EMQ X MQTT Cloud with Home Assistant

In this article, we will introduce to you the process of integrating Home Assistant with EMQ X MQTT Cloud, a Cloud-Native fully managed MQTT service.

Set up EMQ X Cloud MQTT Broker with openHAB

In this article, we will introduce to you the process of integrating openHAB with EMQ X MQTT Cloud, a Cloud-Native fully managed MQTT service.

EMQ X project newsletter 202101

Hello! This is our very first newsletter to share what’s happening in EMQ X open-source team.