//通過cli操作influxdb,(登錄influxdb)
influx -username root -password root
//查看所有數(shù)據(jù)庫(kù)
show databases
//進(jìn)入到某個(gè)數(shù)據(jù)庫(kù)
use influxdb
//查看用戶,兩列數(shù)據(jù)，一列是用戶名稱，一列是是否為管理員用戶
show users
//創(chuàng)建普通戶
create user "influx" with password '123456'
//創(chuàng)建管理員用戶
create user "root" with password '123456' with all privileges
//修改用戶密碼
set password for root= 'root'
/**
* 數(shù)據(jù)庫(kù)設(shè)置admin權(quán)限的兩種方式(建議使用第二種)
*/
GRANT ALL PRIVILEGES ON 數(shù)據(jù)庫(kù)名稱  TO 用戶名稱
GRANT ALL PRIVILEGES TO 用戶名稱
/**
* 數(shù)據(jù)庫(kù)撤銷admin權(quán)限的兩種方式(建議使用第二種)
*/
Revoke ALL PRIVILEGES ON 數(shù)據(jù)庫(kù)名稱 FROM 用戶名稱
Revoke ALL PRIVILEGES FROM 用戶名稱
//查看所有表
show measurements
//查看表的tag
show tag keys from 表名稱
//查看表的field
show field keys from 表名稱
//查看表的series
show series from "表名稱"
//查看當(dāng)前庫(kù)的series
show series

/**
* InfluxDB中沒有顯式的新建表的語句，只能通過insert數(shù)據(jù)的方式來建立新表。
* 其中 cpu_usage 就是表名，resKey索引（tag），value=xx是記錄值（field），索引和記錄值可以有多個(gè)，后面的時(shí)間戳是我們指定的，如果不指定系統(tǒng)自動(dòng)加當(dāng)前時(shí)間時(shí)間戳。
*/
insert cpu_usage resKey=470b14f0-e869-43ed-a8e6-fd634258271f value=0.96516
#指定時(shí)間戳插入
insert cpu_usage resKey=470b14f0-e869-43ed-a8e6-fd634258271f value=0.96516 1688794582731
//刪除表
drop measurement cpu_usage
//刪除數(shù)據(jù)
//influxDB是沒有提供直接刪除數(shù)據(jù)記錄的方法，但是提供數(shù)據(jù)保存策略，主要用于指定數(shù)據(jù)保留時(shí)間，超過指定時(shí)間，就刪除這部分?jǐn)?shù)據(jù)。（數(shù)據(jù)庫(kù)過期策略至少一個(gè)小時(shí)）

//查詢基本構(gòu)成，中括號(hào)[]中間的，都是可選的部分
SELECT * [INTO_clause] FROM "表名" [WHERE_clause] [GROUP_BY_clause] [ORDER_BY_clause] [LIMIT N]

SELECT * FROM "cpu_usage" limit 10 tz('Asia/Shanghai')

SELECT (value+400)*10 FROM "cpu_usage" limit 10 tz('Asia/Shanghai')

支持的操作符：
= 等于
<> 不等于
!= 不等于
> 大于
>= 大于等于
< 小于
<= 小于等于
// 查詢有特定field的key value為字符串的數(shù)據(jù)
SELECT * FROM "cpu_usage" WHERE "resKey" = 'qqqqqq'  tz('Asia/Shanghai')
// 查詢有特定field的key value并且?guī)в?jì)算的數(shù)據(jù)
SELECT (value+400)*10 FROM "cpu_usage" limit 10 tz('Asia/Shanghai')

//查詢10天前的數(shù)據(jù)
SELECT * FROM "cpu_usage" WHERE time > now() - 10d
//根據(jù)指定時(shí)間過濾
SELECT * FROM "cpu_usage" WHERE time >= '2023-06-18T00:00:00Z' AND time <= '2024-08-18T12:30:00Z'

SELECT MEAN("value") FROM "temperature" GROUP BY "field1", "field2"

SELECT MEAN(value) FROM "cpu_usage" WHERE time >= '2023-07-01T10:00:00Z' AND time < '2024-08-02T00:00:00Z' GROUP BY time(30m) tz('Asia/Shanghai')

//上面sql第一條數(shù)據(jù)是從2023-07-01 18:00:00開始的，如果加上offset_interval值為15m, 則查詢第一條數(shù)據(jù)變成了2023-07-01 17:45:00   
SELECT MEAN(value) FROM "cpu_usage" WHERE time >= '2023-07-01T10:00:00Z' AND time < '2024-08-02T00:00:00Z' GROUP BY time(30m,15m) tz('Asia/Shanghai')

//上面sql中，如果需要聚合的時(shí)間段每數(shù)據(jù)，則用fill()中指定的值填充
SELECT MEAN(value) FROM "cpu_usage" WHERE time >= '2023-07-01T10:00:00Z' AND time < '2023-08-02T00:00:00Z' GROUP BY time(30m) fill(10) tz('Asia/Shanghai')

#把表cpu_usage的數(shù)據(jù)導(dǎo)入到備份表cpu_usage_bak中，新表和舊表的tag和Field不變
SELECT * into "cpu_usage_bak" FROM "cpu_usage" 
#該sql只會(huì)把value字段中插入到備份表cpu_usage_bak中，需要插入的字段在原變和舊表中tag和Field不變
SELECT value into "cpu_usage_bak" FROM "cpu_usage" 
#該sql只會(huì)把value和resKey字段中插入到備份表cpu_usage_bak中，需要插入的字段在原變和舊表中tag和Field不變
SELECT value into "cpu_usage_bak" FROM "cpu_usage"  group by resKey
#和上面sql一樣，只不過多個(gè)聚合函數(shù)
SELECT max(value) into "cpu_usage_bak" FROM "cpu_usage"  group by resKey
#該sql插入新表后，字段都會(huì)變成field，如果不想變成Filed，查詢時(shí)加Group By即可
SELECT value,resKey into "cpu_usage_bak" FROM "cpu_usage" 
#該sql中存在子查詢，從子查詢查數(shù)來的數(shù)據(jù)插入到新表，字段都變成了Field
SELECT value,resKey into  "default"."cpu_usage_bak"   FROM  (select * from "default"."cpu_usage" group by resKey)

#求組內(nèi)最大值的平均值
SELECT MEAN("value") FROM (SELECT MAX("value") AS value FROM "cpu_usage" GROUP BY "resKey")
#求兩個(gè)字段差的平均值
SELECT MEAN("difference") FROM (SELECT "字段1" - "字段2" AS "difference" FROM "cpu_usage")

//查看數(shù)據(jù)保存策略(建議使用第二種)
show retention policies on "數(shù)據(jù)庫(kù)名稱"
show retention policies
/**
* 創(chuàng)建數(shù)據(jù)保留策略
* h（小時(shí)），d（天），w（星期）
*/
//CREATE RETENTION POLICY "保留策略名稱" ON "數(shù)據(jù)庫(kù)名稱" DURATION "該保留策略對(duì)應(yīng)的數(shù)據(jù)過期時(shí)間" REPLICATION "復(fù)制因子，開源的InfluxDB單機(jī)環(huán)境永遠(yuǎn)為1" SHARD DURATION "分片組的默認(rèn)時(shí)長(zhǎng)" DEFAULT
//DEFAULT是默認(rèn)保存策略，一個(gè)庫(kù)中只有一個(gè)默認(rèn)保存策略。
create retention policy "testPolice" on myInfluxdb duration 72h    replication 1 SHARD DURATION 1h default
//修改保留策略
alter retention policy "保留策略名稱" on "數(shù)據(jù)庫(kù)名稱" duration 1d
//刪除保留策略
drop retention policy "保留策略名稱" on "數(shù)據(jù)庫(kù)名稱"

//創(chuàng)建分鐘級(jí)別的保存策略：two_hours是策略名稱，rightcloud是庫(kù)名,
CREATE RETENTION POLICY "two_hours" ON "rightcloud" DURATION 2h REPLICATION 1
//創(chuàng)建小時(shí)級(jí)別的保存策略：a_year是策略名稱，rightcloud是庫(kù)名
CREATE RETENTION POLICY "a_year" ON "rightcloud" DURATION 10w REPLICATION 1
//創(chuàng)建連續(xù)查詢：con_query連續(xù)查詢名稱
CREATE CONTINUOUS QUERY "con_query" ON "rightcloud" BEGIN
  SELECT mean("value") AS "value"
  INTO "a_year"."cpu_usage"
  FROM "autogen"."cpu_usage"
  GROUP BY time(10s)
END
//注意:分鐘級(jí)別數(shù)據(jù)和小時(shí)級(jí)別數(shù)據(jù)都是保存在cpu_usage表中,是通過保存策略來區(qū)分開的,在查詢表數(shù)據(jù)時(shí),一定要加上策略名稱
select * from "a_year"."cpu_usage";
select * from "autogen"."cpu_usage";

server:
  port: 8080
spring:
  influx:
    url: http://127.0.0.1:8086 #influxdb服務(wù)器的地址
    username:  #用戶名
    password:  #密碼
    database: rightcloud #指定的數(shù)據(jù)庫(kù)
    retention_policy: autogen

server:  port: 8080spring:  influx:    url: //127.0.0.1:8086 #influxdb服務(wù)器的地址    username:  #用戶名    password:  #密碼    database: rightcloud #指定的數(shù)據(jù)庫(kù)    retention_policy: autogen

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;
/**
 * influxdb配置文件
 */
@Configuration
public class InfluxdbConfig {
    @Value("${spring.influx.url}")
    private String influxDBUrl;
    @Value("${spring.influx.username}")
    private String userName;
    @Value("${spring.influx.password}")
    private String password;
    @Value("${spring.influx.retention_policy}")
    private String retention_policy;
    @Value("${spring.influx.database}")
    private String database;
    @Bean
    public InfluxDB influxdb() {
        InfluxDB influxDB = null;
        if (StringUtils.isEmpty(userName)) {
            influxDB = InfluxDBFactory.connect(influxDBUrl);
        } else {
            influxDB = InfluxDBFactory.connect(influxDBUrl, userName, password);
        }
        try {
            /**
             * 異步插入：
             * enableBatch這里第一個(gè)是point的個(gè)數(shù)，第二個(gè)是時(shí)間，單位毫秒
             * point的個(gè)數(shù)和時(shí)間是聯(lián)合使用的，如果滿100條或者60 * 1000毫秒
             * 滿足任何一個(gè)條件就會(huì)發(fā)送一次寫的請(qǐng)求。
             */
//            influxDB.setDatabase(database).enableBatch(100, 1000 * 60, TimeUnit.MILLISECONDS);
            influxDB.setDatabase(database);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //設(shè)置默認(rèn)策略
            this.retention_policy = retention_policy == null || "".equals(retention_policy) ? "autogen" : retention_policy;
            influxDB.setRetentionPolicy(retention_policy);
        }
        //設(shè)置日志輸出級(jí)別
        influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
        return influxDB;
    }
}

import org.influxdb.InfluxDB;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
 *
 */
@Component
public class InfluxDBUtils {
    @Autowired
    private InfluxDB influxDB;
    @Value("${spring.influx.database}")
    private String database;
    /**
     * 插入單條數(shù)據(jù)寫法1
     *
     * @param measurement
     */
    public void insertOne01(String measurement) {
        //構(gòu)建
        Point.Builder builder = Point.measurement(measurement);
        //可指定時(shí)間戳
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        //tag屬性只能存儲(chǔ)String類型
        builder.tag("name", "zhanggang");
        //設(shè)置field
        builder.addField("filed", "fileValue");
        influxDB.write(builder.build());
    }
    /**
     * 插入單條數(shù)據(jù)寫法2
     *
     * @param measurement
     */
    public void insertOne02(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        //構(gòu)建
        Point.Builder builder = Point.measurement(measurement);
        //可指定時(shí)間戳
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        //tag屬性只能存儲(chǔ)String類型
        builder.tag(tags);
        //設(shè)置field
        builder.fields(fields);
        influxDB.write(builder.build());
    }
    /**
     * 插入單條數(shù)據(jù)
     * influxDB開啟UDP功能, 默認(rèn)端口:8089,默認(rèn)數(shù)據(jù)庫(kù):udp,沒提供代碼傳數(shù)據(jù)庫(kù)功能接口
     * 使用UDP的原因
     * TCP數(shù)據(jù)傳輸慢，UDP數(shù)據(jù)傳輸快。
     * 網(wǎng)絡(luò)帶寬需求較小，而實(shí)時(shí)性要求高。
     * InfluxDB和服務(wù)器在同機(jī)房，發(fā)生數(shù)據(jù)丟包的可能性較小，即使真的發(fā)生丟包，對(duì)整個(gè)請(qǐng)求流量的收集影響也較小。
     *
     * @param measurement
     * @param tags
     * @param fields
     */
    public void insertUDPOne03(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        //構(gòu)建
        Point.Builder builder = Point.measurement(measurement);
        //可指定時(shí)間戳
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        //tag屬性只能存儲(chǔ)String類型
        builder.tag(tags);
        //設(shè)置field
        builder.fields(fields);
        int udpPort = 8089;
        influxDB.write(udpPort, builder.build());
    }
    //批量插入1
    public void insertBatchByRecords() {
        List<String> lines = new ArrayList<String>();
        String measurement = "test-inset-one";
        for (int i = 0; i < 2; i++) {
            Point.Builder builder = Point.measurement(measurement);
            //tag屬性只能存儲(chǔ)String類型
            builder.tag("name", "zhanggang" + i);
            //設(shè)置field
            builder.addField("filed", "fileValue" + i);
            lines.add(builder.build().lineProtocol());
        }
        influxDB.write(lines);
    }
    //批量插入2
    public void insertBatchByPoints() {
        BatchPoints batchPoints = BatchPoints.database(database)
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        String measurement = "test-inset-one";
        for (int i = 0; i < 2; i++) {
            Point.Builder builder = Point.measurement(measurement);
            //tag屬性只能存儲(chǔ)String類型
            builder.tag("name", "zhanggang" + i);
            //設(shè)置field
            builder.addField("filed", "fileValue" + i);
            batchPoints.point(builder.build());
        }
        influxDB.write(batchPoints);
    }
    /**
     * 查詢，返回Map集合
     *
     * @param query 完整的查詢語句
     * @return
     */
    public List<Map<String, Object>> fetchRecords(String query) {
        List<Map<String, Object>> results = new ArrayList<>();
        QueryResult queryResult = influxDB.query(new Query(query, database));
        queryResult.getResults().forEach(result -> {
            result.getSeries().forEach(serial -> {
                List<String> columns = serial.getColumns();
                int fieldSize = columns.size();
                serial.getValues().forEach(value -> {
                    Map<String, Object> obj = new HashMap<String, Object>();
                    for (int i = 0; i < fieldSize; i++) {
                        obj.put(columns.get(i), value.get(i));
                    }
                    results.add(obj);
                });
            });
        });
        return results;
    }
    /**
     * 批量寫入數(shù)據(jù)
     *
     * @param database        數(shù)據(jù)庫(kù)
     * @param retentionPolicy 保存策略
     * @param consistency     一致性
     * @param records         要保存的數(shù)據(jù)（調(diào)用BatchPoints.lineProtocol()可得到一條record）
     */
    public void batchInsert(final String database, final String retentionPolicy,
                            final InfluxDB.ConsistencyLevel consistency, final List<String> records) {
        influxDB.write(database, retentionPolicy, consistency, records);
    }
    /**
     * 查詢
     *
     * @param command 查詢語句
     * @return
     */
    public QueryResult query(String command) {
        return influxDB.query(new Query(command, database));
    }
    /**
     * 創(chuàng)建數(shù)據(jù)保留策略
     * 設(shè)置數(shù)據(jù)保存策略 defalut 策略名 /database 數(shù)據(jù)庫(kù)名/ 30d 數(shù)據(jù)保存時(shí)限30天/ 1 副本個(gè)數(shù)為1/ 結(jié)尾DEFAULT
     * 表示 設(shè)為默認(rèn)的策略
     */
    public void createRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "defalut", database, "30d", 1);
        this.query(command);
    }
 }

@RestController
public class InfluxDBController {
    @Autowired
    private InfluxDBUtils influxDBUtils;
    @GetMapping("/insertOne")
    public String insertOne() {
        String measurement = "test-inset-one";
        influxDBUtils.insertOne01(measurement);
        return "success";
    }
    @GetMapping("/query01")
    public List<Map<String, Object>> query01() {
        String sql = "SELECT * FROM \"test-inset-one\"   TZ('Asia/Shanghai')";
        return influxDBUtils.fetchRecords(sql);
    }