前言

之前使用Rabbitmq部署了一個簡單的MQTT伺服器,暫未做使用者隔離,也部署了InfluxDB時序資料庫,但是並不能直接通過配置將MQTT伺服器的資料轉存至時序資料庫中,於是我決定自己寫指令碼實現下.

準備

開啟shell使用
pip install influxdb安裝InluxDB所需模組
pip install paho-mqtt安裝Rabbmq所需模組

原始碼

# coding=utf-8
import json
import random
import threading
import os
import paho.mqtt.client as mqtt
import
 
 time
from influxdb import InfluxDBClient
from my_lib.code_handle.code_handle import auto_code


class Mqtt_handle:
    topic_sub='$dp'
    topic_pub='$info'
    counts = 0
    clientID = ''
    for i in range(0, 2):
        clientID = clientID.join(str(random.uniform(0, 1)))
    mqtt_client = mqtt.Client(clientID)
    DB_client = InfluxDBClient(self._host, 8086
 
, '', '', 'mydb')  # 初始化

    def __init__(self, host, port):
        self._host = host
        self._port = port
        self.mqtt_client.on_connect = self._on_connect  # 設定連線上伺服器回撥函式
        self.mqtt_client.on_message = self._on_message  # 設定接收到伺服器訊息回撥函式

    def connect(self, username=None, password=None)
 
:
        self.mqtt_client.username_pw_set(username, password)
        self.mqtt_client.connect(self._host, self._port, 60)  # 連線伺服器,埠為1883,維持心跳為60秒

    def publish(self, data):
        self.mqtt_client.publish(self.topic_pub, data)

    def loop(self, timeout=None):
        thread = threading.Thread(target=self._loop, args=(timeout,))
        thread.start()

    def _loop(self, timeout=None):
        if not timeout:
            self.mqtt_client.loop_forever()
        else:
            self.mqtt_client.loop(timeout)

    def _on_connect(self, client, userdata, flags, rc):
        local_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
        with open('./run.log', 'a+')as f:
            f.write('@Run ' + local_time + ' Connected with result code :' + str(rc))
        client.subscribe(self.topic_sub)

    def _on_message(self, client, userdata, msg):  # 從伺服器接受到訊息後回撥此函式
        data_json = auto_code(str(msg.payload))
        if self._is_json(data_json):
            data_list = [json.loads(data_json)]
            #如果符合InfluxDB格式就轉存至資料庫
            if 'measurement' in data_list[0] and 'tags' in data_list[0] and 'fields' in data_list[0]:
                try:
                    DB_client.write_points(data_list)
                    self.counts += 1
                    local_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
                    #//**********//記錄一個上傳日誌
                    with open('./upload.log', 'a+')as f:
                        f.write('Success,counts:' + str(self.counts) + ' Time:' + local_time + '\n')
                except Exception as e:
                    with open('./upload.log', 'a+')as f:
                        f.write(e.message)
                        f.write('\nTopic:' + auto_code(str(msg.topic)) + " Msg:" + data_json + '\n\n')
                    #//**********//
            #如果接受到停止指令就停止程式並記錄一個停止日誌
            elif data_list[0].has_key('cmd') and data_list[0]['cmd'] == 'exit':
                print '\[email protected]_handle Exit\n'
                with open('./run.log', 'a')as f:
                    local_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
                    f.write(
                        '@Stop ' + local_time + ' Topic:' + auto_code(str(msg.topic)) + " Msg:" + data_json + '\n')
                os._exit(0)#停止程式
    #解析JSON前先判斷資料 是否是JSON格式,避免程式崩潰
    def _is_json(self, data):
        try:
            json.loads(data)
        except ValueError:
            return False
        return True


if __name__ == '__main__':
    local_host = '127.0.0.1'
    DB_client = InfluxDBClient(local_host, 8086, '', '', 'mydb')  # 初始化
    mqtt_client = Mqtt_handle(local_host, 1883)
    mqtt_client.connect('influxdb', 'influxdb')
    mqtt_client.loop()

寫一個簡單的Shell執行一下指令碼:

#!/bin/bash
nohup python /home/ubuntu/app/py/mqttDB/mqtt_handle.py &