MQTT客户端信息

2022-02-07

字数统计: 644字 | 阅读时长: 3分

文章导航

保存客户端的信息，用来显示客户端的状态、实时值

在mqtt数据库中先建mqtt_client，用来保存客户端信息

#之前新建的mqtt_user表
CREATE TABLE `mqtt_user` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `username` varchar(100) DEFAULT NULL,
  `password` varchar(100) DEFAULT NULL,
  `salt` varchar(35) DEFAULT NULL,
  `is_superuser` tinyint(1) DEFAULT 0,
  `created` datetime DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `mqtt_username` (`username`)
) ENGINE=MyISAM DEFAULT CHARSET=utf8;

#新建mqtt_client表
CREATE TABLE `mqtt_client` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `client_id` varchar(100) DEFAULT NULL COMMENT 'client_id',
  `user_id` INT(11) unsigned,
  `username` varchar(100) DEFAULT NULL,
  `real-time_value` varchar(100) DEFAULT NULL COMMENT '实时值,比如实时温度',
  `is_online` int(1) DEFAULT NULL COMMENT '0: off line, 1: on line',
  `topic` varchar(100) NOT NULL DEFAULT '' COMMENT 'topic',
  `created_at` DATETIME COMMENT 'create time',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB comment  = '客户端信息' DEFAULT CHARSET=utf8;

# `clientId`唯一值
ALTER TABLE `mqtt_client` ADD UNIQUE( `clientId`);

# 外键
Alter table `mqtt_client` add constraint `u_id` foreign key(`user_id`) references `mqtt_user`(`id`) ON DELETE CASCADE ON UPDATE CASCADE;

设备将内容(例如){“clientId”:”xxx”,”real-time_value”:”38度”}发送到EMQX的主题real-time_value,然后webhook到服务器，保存实时值.

设置emqx_web_hook.conf，开启EMQX的emqx_web_hook插件

# 5uks.net
web.hook.url = http://127.0.0.1:8082/mqtt/webhook
web.hook.rule.client.connected.1     = {"action": "on_client_connected"}               
web.hook.rule.client.disconnected.1  = {"action": "on_client_disconnected"}               
web.hook.rule.message.publish.1      = {"action": "on_message_publish"}

用Flask接收,并保存到数据库.

from flask import Flask, render_template, request
from json import dumps,loads
from pprint import pprint

app = Flask(__name__)
# EMQX WebHook
@app.route('/mqtt/webhook',methods=['GET','POST'])
def mqtt_webhook():
    if request.method == 'POST':
        data=request.json
        print(data)
        action = data['action']  # action
        # clientid = data['clientid']  # clientid
        res={
            'action':action,
            # 'clientid':clientid
        }
        # pprint(res)
        return dumps(res,ensure_ascii=False)
    else:
        return "只能使用POST方式访问本接口",401

if __name__ == '__main__':
    from gevent import pywsgi
    # host='127.0.0.1' #只有本地访问
    host = '0.0.0.0'  # 任何主机都可以访问
    port = 8082
    # server = pywsgi.WSGIServer((host, port), app)
    # print('server start')
    # server.serve_forever()
    app.run(host=host, port=port, debug=True)

# `clientId`唯一值
ALTER TABLE `mqtt_client` ADD UNIQUE( `clientId`);


# 不存在则插入，存在则更新
INSERT INTO TABLE (a,c) VALUES (1,3) ON DUPLICATE KEY UPDATE c=VALUES(c);

INSERT INTO `mqtt_client` (`clientId`,`isOnline`) VALUES ('mqttdash-315fc1b5',1) ON DUPLICATE KEY UPDATE `isOnline`=VALUES(`isOnline`)

# mysql 在值为null的字段插入数据

# 如果state的数据类型是 int 等数子类型或者bit类型:
update 表 set state=新数字 where state is null

update `mqtt_client` set `createdAt` ='2017-07-25 19:38:14' where `createdAt`  is null and `clientId`='mqttdash-315fc1b5';

# Python3代码
import time

# 输入毫秒级的时间，转出正常格式的时间
def timeStamp(timeNum):
    timeStamp = float(timeNum/1000)
    timeArray = time.localtime(timeStamp)
    otherStyleTime = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
    print (otherStyleTime)
    return otherStyleTime