物联网架构成长之路(33)-EMQ数据存储到influxDB

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

1 [http]
2     enabled = true
3     bind-address = ":8086"
4     auth-enabled = false
5     log-enabled = true
6     write-tracing = false
7     pprof-enabled = true

1 --显示所有用户：
2 show users
3 --新增用户：
4 --普通用户 (注意：用户名用双引号，密码用单引号)
5 create user "user" with password ‘user‘
6 --管理员用户
7 create user "admin" with password ‘admin‘ with all privileges
8 --删除用户
9 drop user "user"

1 --创建数据库
2 create database wunaozai
3 --创建好后，就可以不用管了。一些简单的操作，可以参考其他博客资料。
4 --删除数据库
5 drop database wunaozai
6 --切换使用数据库
7 use wunaozai

1 --显示所有表
2 show measurements
3 --新建表(往表里面插入数据，就是新建表了)
4 --插入数据的语法有点特殊，采用的是InfluxDB特有的语法：
5 <measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]
6 insert wnztable,tag=mqtt value=33
7 --删除表
8 drop measurements wunaozai

1 <!-- https://mvnrepository.com/artifact/org.influxdb/influxdb-java -->
2 <dependency>
3 <groupId>org.influxdb</groupId>
4 <artifactId>influxdb-java</artifactId>
5 <version>2.15</version>
6 </dependency>

 1 import org.influxdb.InfluxDB;
 2 import org.influxdb.InfluxDBFactory;
 3 import org.influxdb.dto.Point;
 4 
 5 /**
 6  * 数据缓存至InfluxDB
 7  * @author wunaozai
 8  *
 9  */
10 public class InfluxDBService {
11     
12     private static String INFLUXDB_URL = "http://127.0.0.1:8086";
13     private static String INFLUXDB_USERNAME = "admin";
14     private static String INFLUXDB_PASSWORD = "admin";
15     private static String INFLUXDB_DATABASE = "wunaozai"; //注意这里对应数据库，一般要先在命令行中创建数据库
16     private static InfluxDB influxDB = null;
17     
18     private InfluxDBService(){
19         
20     }
21     
22     public static InfluxDB getInstance(){
23         if(influxDB == null){
24             influxDB = InfluxDBFactory.connect(INFLUXDB_URL, INFLUXDB_USERNAME, INFLUXDB_PASSWORD);
25             influxDB.setDatabase(INFLUXDB_DATABASE);
26             influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
27         }
28         return influxDB;
29     }
30     public static int writePoint(Point point){
31         getInstance().write(point);
32         return 0;
33     }
34 }

public void messageArrived(String topic, MqttMessage message);

 1     @Override
 2     public void messageArrived(String topic, MqttMessage message) throws Exception {
 3         try{
 4             System.out.println(topic);
 5             String json = new String(message.getPayload());
 6             MQTTProtocolVoModel protocol = BaseModel.parseJSON(json, MQTTProtocolVoModel.class);
 7             
 8             String cmd = protocol.getCmd();
 9             String customer_id = protocol.getProfile().getCustomer_id(); //厂商ID
10             String product_id = protocol.getProfile().getProduct_id(); //产品ID
11             String device_sn = protocol.getProfile().getDevice_sn(); //设备ID
12             Map<String, String> para = protocol.getDatapoint().getPara();
13             Map<String, Object> fields = new HashMap<>(); //这里是客户端传过来的数据点，就是需要被显示和监控的数据
14             for (Map.Entry<String, String> entry : para.entrySet()) {
15                 fields.put(entry.getKey(), entry.getValue());
16             }
17             Map<String, String> tag = new HashMap<>();
18             tag.put("customer_id", customer_id);
19             tag.put("product_id", product_id);
20             tag.put("device_sn", device_sn);
21             //这里可以添加很多Tag，为了简单演示，这里隐藏部分Tag
22             //构造数据点
23             Point point = Point.measurement("datapoint")
24                     .tag(tag).fields(fields).build();
25             InfluxDBService.writePoint(point);
26         }catch (Exception e) {
27             e.printStackTrace();
28         }
29     }