物聯網架構成長之路(33)-EMQ數據存儲到influxDB

一、前言
　　時隔一年半，技術變化特別快，學習也要跟上才行。以前寫過EMQ數據轉存問題，當時用了比較笨的方法，通過寫插件的方式，把MQTT里面的數據發送到數據庫進行存儲。當時也是為了學習erlang和emq。現在隨着對物聯網的深入，也結合實際需求，不停的學習。
下面將介紹我實驗測試可行的物聯網數據分析解決方案。采用的還是開源方案。通過訂閱MQTT的根Topic，把所有物聯網數據轉存到InfluxDB時序數據庫，然后通過Grafana進行圖表顯示。這應該是目前比較流行的方案。
二、安裝InfluxDB
　　InfluxDB是時序數據庫，特別適合做數據監控和物聯網數據存儲。【也可以說適合我現在參與架構的物聯網平台的技術選型】
　　針對InfluxDB也沒有什么可以多說的，詳細可以查閱官方文檔，或者網上的博客文章。我寫的都是平時實踐過程的操作記錄，寫博客，主要是為了以后忘記的時候，回看查閱用的。另一方面是加強跟同行讀者交流的渠道。有一點要注意，一開始為了新，我用InfluxDB 2.0 版本，發現不行，那個太新的，很多對應的開發庫沒有完善好。所以還是采用InfluxDB 1.x版本。這樣在spring boot 里面也有自帶的starter庫可以使用，操作起來特別方便。
　　InfluxDB官方文檔: https://docs.influxdata.com/influxdb/v1.7/   安裝:

wget -qO- https://repos.influxdata.com/influxdb.key | sudo apt-key add -
echo "deb https://repos.influxdata.com/debian stretch stable" | sudo tee /etc/apt/sources.list.d/influxdb.list
apt-get update
apt-get install influxdb

 

三、InfluxDB基礎命令使用
　　修改配置文件 /etc/influxdb/influxdb.conf

[http]
    enabled = true
    bind-address = ":8086"
    auth-enabled = false
    log-enabled = true
    write-tracing = false
    pprof-enabled = true

　　這里先設置不授權，等一下創建用戶后，再修改為 auth-enabled=true，這個一般也是屬於內部應用，不用ssl加密了。即使要也是通過Nginx進行反向代理。

　　用戶管理

--顯示所有用戶：
show users
--新增用戶：
--普通用戶 (注意：用戶名用雙引號，密碼用單引號)
create user "user" with password 'user'
--管理員用戶
create user "admin" with password 'admin' with all privileges
--刪除用戶
drop user "user"

　　創建好后，注意修改influxdb.conf 中的 auth-enable=true， 然后重啟服務 service influxdb restart

--創建數據庫
create database wunaozai
--創建好后，就可以不用管了。一些簡單的操作，可以參考其他博客資料。
--刪除數據庫
drop database wunaozai
--切換使用數據庫
use wunaozai

--顯示所有表
show measurements
--新建表(往表里面插入數據，就是新建表了)
--插入數據的語法有點特殊，采用的是InfluxDB特有的語法：
<measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]
insert wnztable,tag=mqtt value=33
--刪除表
drop measurements wunaozai

　　其他的高級語法，不如查詢還有策略就不展開，暫時不是重點，等以后深入研究后，在寫博客介紹。

四、EMQ轉存InfluxDB
　　EMQ如何把消息轉存到InfluxDB呢，就是本章節的重點，利用上一篇博客中提到的，SpringBoot客戶端監聽EMQ的根Topic，然后把需要進行轉存的Topic及其對應的Payload，構造成InfluxDB表數據，然后插入到InfluxDB中。
　　下面介紹一下用到的InfluxDB工具類
　　先在pom.xml中引入InfluxDB相關jar包

<!-- https://mvnrepository.com/artifact/org.influxdb/influxdb-java -->
<dependency>
<groupId>org.influxdb</groupId>
<artifactId>influxdb-java</artifactId>
<version>2.15</version>
</dependency>

　　相關工具類代碼

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;

/**
 * 數據緩存至InfluxDB
 * @author wunaozai
 *
 */
public class InfluxDBService {
    
    private static String INFLUXDB_URL = "http://127.0.0.1:8086";
    private static String INFLUXDB_USERNAME = "admin";
    private static String INFLUXDB_PASSWORD = "admin";
    private static String INFLUXDB_DATABASE = "wunaozai"; //注意這里對應數據庫，一般要先在命令行中創建數據庫
    private static InfluxDB influxDB = null;
    
    private InfluxDBService(){
        
    }
    
    public static InfluxDB getInstance(){
        if(influxDB == null){
            influxDB = InfluxDBFactory.connect(INFLUXDB_URL, INFLUXDB_USERNAME, INFLUXDB_PASSWORD);
            influxDB.setDatabase(INFLUXDB_DATABASE);
            influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
        }
        return influxDB;
    }
    public static int writePoint(Point point){
        getInstance().write(point);
        return 0;
    }
}

　　在上一篇博客中的MqttPushCallback.java中的

public void messageArrived(String topic, MqttMessage message);

　　這個函數來轉存。

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        try{
            System.out.println(topic);
            String json = new String(message.getPayload());
            MQTTProtocolVoModel protocol = BaseModel.parseJSON(json, MQTTProtocolVoModel.class);
            
            String cmd = protocol.getCmd();
            String customer_id = protocol.getProfile().getCustomer_id(); //廠商ID
            String product_id = protocol.getProfile().getProduct_id(); //產品ID
            String device_sn = protocol.getProfile().getDevice_sn(); //設備ID
            Map<String, String> para = protocol.getDatapoint().getPara();
            Map<String, Object> fields = new HashMap<>(); //這里是客戶端傳過來的數據點，就是需要被顯示和監控的數據
            for (Map.Entry<String, String> entry : para.entrySet()) {
                fields.put(entry.getKey(), entry.getValue());
            }
            Map<String, String> tag = new HashMap<>();
            tag.put("customer_id", customer_id);
            tag.put("product_id", product_id);
            tag.put("device_sn", device_sn);
            //這里可以添加很多Tag，為了簡單演示，這里隱藏部分Tag
            //構造數據點
            Point point = Point.measurement("datapoint")
                    .tag(tag).fields(fields).build();
            InfluxDBService.writePoint(point);
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

　　這里可以通過EMQ Dashboard自帶的Websocket進行發送，也可以通過前面小節用到的PC工具，網上Web端MQTT客戶也很多，可以通過任意MQTT工具進行測試。
　　下面這個是查詢InfluxDB得到的表數據。

 

參考資料：
　　https://www.cnblogs.com/jason1990/p/11076310.html
　　https://blog.csdn.net/caodanwang/article/details/51967393
　　https://docs.influxdata.com/influxdb/v1.7/
　　https://www.cnblogs.com/shhnwangjian/p/6897216.html

本文地址: https://www.cnblogs.com/wunaozai/p/11160730.html