基于DockerSwarm 部署InfluxDB并使用JAVA操作

Docker中部署InfluxDB

1、運作容器

$ docker run --rm \
      -e INFLUXDB_DB=db0 -e INFLUXDB_ADMIN_ENABLED=true \
      -e INFLUXDB_ADMIN_USER=admin -e INFLUXDB_ADMIN_PASSWORD=supersecretpassword \
      -e INFLUXDB_USER=telegraf -e INFLUXDB_USER_PASSWORD=secretpassword \
      -v $PWD:/var/lib/influxdb \
      influxdb /init-influxdb.sh      

2、Stack部署

（1）先建立一個Config，名稱為influxdb.conf，内容如下：

[meta]
dir = "/var/lib/influxdb/meta"
retention-autocreate = true
logging-enabled = true

[data]
dir = "/var/lib/influxdb/data"
index-version = "inmem"
wal-dir = "/var/lib/influxdb/wal"
wal-fsync-delay = "0s"
query-log-enabled = true
cache-max-memory-size = 1073741824
cache-snapshot-memory-size = 26214400
cache-snapshot-write-cold-duration = "10m0s"
compact-full-write-cold-duration = "4h0m0s"
max-series-per-database = 1000000
max-values-per-tag = 100000
max-concurrent-compactions = 0
trace-logging-enabled = false
[http]
enabled = true
bind-address = ":8086"
auth-enabled = false
log-enabled = true
write-tracing = false
pprof-enabled = true
https-enabled = false
https-certificate = "/etc/ssl/influxdb.pem"
https-private-key = ""
max-row-limit = 0
max-connection-limit = 0
shared-secret = ""
realm = "InfluxDB"
unix-socket-enabled = false
bind-socket = "/var/run/influxdb.sock"      

(2) docker swarm角本
      

version: '3.3'
services:
  influxdb:
    image: influxdb:1.2.0
    ports:
      - 8086:8086
      - 8083:8083
      - 2003:2003
    environment:
      INFLUXDB_DB: db0
      INFLUXDB_ADMIN_ENABLED: 1
      INFLUXDB_ADMIN_USER: admin
      INFLUXDB_ADMIN_PASSWORD: admin
      INFLUXDB_USER: user
      INFLUXDB_USER_PASSWORD: user
    volumes:
      - /opt/docker/influxdb/data:/var/lib/influxdb
    configs:
      - source: influxdb.conf
        target: /etc/influxdb/influxdb.conf
configs:
  influxdb.conf:
    external: true      

說明：InfluxDB自1.2.0之後，取消了WEB管理頁面。是以我們部署的是1.2.0版本，部署完後可以通路：http://<ip>:8083來通路管理端。

      

3、使用JAVA插入測試

（1）引入Maven包：

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

（2）編寫測試插入代碼（網上粘一段）：

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
import org.influxdb.dto.Point.Builder;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

import java.util.Map;

/**
 * 時序資料庫 InfluxDB 連接配接
*/
public class InfluxDBConnect {

    private String username;
    private String password;
    private String url;
    private String database;

    private InfluxDB influxDB;


    public InfluxDBConnect(String username, String password, String url, String database) {
        this.username = username;
        this.password = password;
        this.url = url;
        this.database = database;
    }

    /** 連接配接時序資料庫；獲得InfluxDB **/
    public InfluxDB connection() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(url, username, password);
        }
        return influxDB;
    }

    /**
     * 設定資料儲存政策
     * defalut 政策名 /database 資料庫名/ 30d 資料儲存時限30天/ 1  副本個數為1/ 結尾DEFAULT 表示 設為預設的政策
     */
    public void createRetentionPolicy(int retentionDay, int replicationCount) {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "defalut", database, retentionDay + "d", replicationCount);
        this.query(command);
    }

    /**
     * 查詢
     * @param command 查詢語句
     * @return
     */
    public QueryResult query(String command) {
        return influxDB.query(new Query(command, database));
    }

    /**
     * 插入
     * @param measurement 表
     * @param tags        标簽
     * @param fields      字段
     */
    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(fields);

        influxDB.write(database, "", builder.build());
    }

    /**
     * 删除
     * @param command 删除語句
     * @return 傳回錯誤資訊
     */
    public String deleteMeasurementData(String command) {
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /**
     * 建立資料庫
     * @param dbName 庫名稱
     */
    public void createDB(String dbName) {
        this.query("create database " + dbName);
    }

    /**
     * 删除資料庫
     * @param dbName
     */
    public void deleteDB(String dbName) {
        this.query("drop database " + dbName);
    }      

/**
     * 插入
    */
    public void insert(InfluxDbRow influxDbRow) {
    if (influxDbRow == null) {
        return;
    }
    Point.Builder builder = Point.measurement(influxDbRow.getMeasurement());
    builder.tag(influxDbRow.getTags());
    builder.fields(influxDbRow.getFields());
    if (influxDbRow.getTimeSecond() != null) {
        builder.time(influxDbRow.getTimeSecond(), TimeUnit.SECONDS);
    }
    influxDB.write(database, "default", builder.build());
   }      

}      

（3）調用插入監控資料

      

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.util.TypeUtils;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;

/**
 * @author song 2019/8/27 16:28
 */
public class InsertTest {
    private InfluxDBConnect influxdb;
    private String username = "admin";//使用者名
    private String password = "admin";//密碼
    private String openurl = "http://54.222.182.173:8086";//連接配接位址
    private String database = "db0";//資料庫
    private String measurement = "sys_code";

    @Before
    public void setUp() {
        influxdb = new InfluxDBConnect(username, password, openurl, database);
        influxdb.connection();
        influxdb.createRetentionPolicy(30, 1);
    }

    @Test
    public void testInsert() throws IOException {
        String data = new String(Files.readAllBytes(Paths.get("C:\\Aliyun\\執行個體監控資料.json")));
        JSONObject jobj = JSON.parseObject(data);
        JSONArray array = jobj.getJSONObject("MonitorData").getJSONArray("InstanceMonitorData");
        for (Object o : array) {
            JSONObject obj = (JSONObject) o;
            Map<String, Object> fields = new HashMap<>();
            long time = TypeUtils.castToDate(obj.get("TimeStamp")).getTime() / 1000;
            fields.put("TimeStamp", time);
            fields.put("IOPSRead", obj.get("IOPSRead"));
            fields.put("IOPSWrite", obj.get("IOPSWrite"));
            fields.put("IntranetBandwidth", obj.get("IntranetBandwidth"));
            fields.put("BPSRead", obj.get("BPSRead"));
            fields.put("BPSWrite", obj.get("BPSWrite"));
            fields.put("IntranetTX", obj.get("IntranetTX"));
            fields.put("IntranetRX", obj.get("IntranetRX"));
            fields.put("CPU", obj.get("CPU"));
            fields.put("InternetRX", obj.get("InternetRX"));
            fields.put("InternetTX", obj.get("InternetTX"));
            Map<String, String> tags = new HashMap<>();
            tags.put("InstanceId", obj.get("InstanceId").toString());
            influxdb.insert("ecs_time2", tags, fields);
        }
    }

}      

（4）效果：

按時間的查詢，可以是：SELECT * FROM "aliyun_ecs" WHERE time >'2019-08-27T14:39:00Z' and time < '2019-08-27T15:39:00Z'      

宋興柱(Sindrol)：轉載内容，請标明出處，謝謝！源文來自 寶貝雲知識分享：https://www.dearcloud.cn