Influxdb和TDengine的写入性能测试（Java）

这里读写测试是为了大致了解下Influxdb和TDengine的读写速度，以及对比数据存储大小。Influxdb是用go语言开发的，TDengine是用c语言开发，Influxdb和TDengine都支持单机版和集群版，这里都使用单机版测试。TDengine服务端下载地址TDengine-server-2.4.0.12-Linux-x64.tar.gz (15.5 M)TDengine Windows客户端下载地址TDengine-client-2.4.0.12-Windows-x86.exeTDengine 官方文档

TDengine提供两种读写方式，这里我们使用JNI方式。



这张图是之前用influxdata公司提供的工具测试的
一、最终测试结果

有一点需要注意，从后面我贴出的源码也可以看出，写入Influxdb的数据是在计时前准备好的，并且数据都是已经序列化过的数据；而TDengine的数据是边组织边写入，这在一定程度上也拖慢了写入速度，不过开篇我已经说了，我这里旨在大概了解下，不做科学准确的性能对比测试哈（TDengine官方的测试对比更科学可靠）。无论是写入还是查询，Influxdb的cpu负载和内存占用会随着数据量的增加而增加，例如一亿条数要5秒左右出结果，而TDengine查询几乎秒出，cpu负载和内存占用都较低且波动不大。

maven 依赖

		
            com.taosdata.jdbc
            taos-jdbcdriver
            2.0.37
        

package com.cloudansys.test;

import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.TSDBPreparedStatement;
import okhttp3.OkHttpClient;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class TestWriteRead {

    public static void main(String[] args) throws Exception {
        int factor = 2000;
        writeToTDengine(factor);
//        writeToInfluxDB(factor);
//        queryTDengine();
    }

    
    private static void writeToTDengine(int factor) throws Exception {
        Connection conn = getTdConn();
        int numOfTid = 100, ef = factor / 100, numOfRow = ef * 10000;
        int bf = 2000, count = numOfRow / bf, left = factor;
        String sql = "insert into ? using st_wy tags(?) values(?,?,?,?)";
        TSDBPreparedStatement pStmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class);
        long startTime = System.currentTimeMillis();
        System.out.println("=======================<写入开始>=======================");
        for (int i = 1; i <= numOfTid; i++) {
            long current = System.currentTimeMillis();
            for (int j = 0; j < count; j++) {
                // set table name
                pStmt.setTableName("t_" + i);
                // set tags
                pStmt.setTagInt(0, i);
                // set columns
                ArrayList tsList = new ArrayList<>();
                for (int k = 0; k < bf; k++) {
                    tsList.add(current++);
                }
                // set timestamp
                pStmt.setTimestamp(0, tsList);
                // set v1 v2 v3
                pStmt.setDouble(1, ranValues(bf));
                pStmt.setDouble(2, ranValues(bf));
                pStmt.setDouble(3, ranValues(bf));
                // add column
                pStmt.columnDataAddBatch();
                // execute column
                pStmt.columnDataExecuteBatch();
            }
            left -= ef;
            System.out.print("r" + factor + "万条数据写入中：" + left + " 万条");
        }
        System.out.println("n=======================<写入结束>=======================");
        long endTime = System.currentTimeMillis();
        System.out.println("总共耗时：" + (endTime - startTime) / 1000 + "s");
        conn.close();
    }

    
    private static void initSchema(Connection conn) throws Exception {
        Statement stmt = conn.createStatement();
        stmt.execute("create stable st_wy(ts timestamp, v1 double, v2 double, v3 double) tags(tid int);");
    }

    
    private static void queryTDengine() throws Exception {
        Connection conn = getTdConn();
        Statement stmt = conn.createStatement();
        String sql = "select * from t_1 limit 5;";
        ResultSet resultSet = stmt.executeQuery(sql);
        Timestamp ts;
        double v1, v2, v3;
        System.out.println("ts-v1-v2-v3");
        String template = "{}-{}-{}-{}";
        while (resultSet.next()) {
            ts = resultSet.getTimestamp(1);
            v1 = resultSet.getDouble("v1");
            v2 = resultSet.getDouble(3);
            v3 = resultSet.getDouble(4);
            System.out.println(StrUtil.format(template, ts, v1, v2, v3));
        }
        conn.close();
    }

    
    private static ArrayList ranValues(int size) {
        ArrayList resList = new ArrayList<>();
        for (int i = 0; i < size; i++) {
            resList.add(NumberUtil.round(Math.random(), 3).doublevalue());
        }
        return resList;
    }

    
    private static void writeToInfluxDB(int factor) {
        List records = getRecords(factor);
        InfluxDB influxDB = getInfluxDB();
        long startTime = System.currentTimeMillis();
        System.out.println("n=======================<写入开始>=======================");
        writeRecords(influxDB, records);
        System.out.println("n=======================<写入结束>=======================");
        long endTime = System.currentTimeMillis();
        System.out.println("总共耗时：" + (endTime - startTime) / 1000 + "s");
        influxDB.close();
    }

    
    private static void writeRecords(InfluxDB influxDB, List records) {
        List dataList = new ArrayList<>();
        int num = 0, batchFactor = 25, factor = records.size() / 10000;
        for (String record : records) {
            dataList.add(record);
            if (dataList.size() == batchFactor * 10000) {
                influxDB.write(dataList);
                dataList.clear();
                System.out.print("r" + factor + "万条数据写入中：" + (factor - batchFactor * ++num) + " 万条");
            }
        }
    }

    
    private static List getRecords(int factor) {
        List records = new ArrayList<>();
        Point.Builder pointBuilder = Point.measurement("wy");
        long curMillis = System.currentTimeMillis();
        int num = factor / 100, count = num * 10000;
        for (int i = 1; i <= 100; i++) {
            System.out.print("r" + factor + "万条数据准备中：" + i * num + " 万条");
            for (int j = 1; j <= count; j++) {
                pointBuilder
                        .time(curMillis++, TimeUnit.MILLISECONDS)
                        .tag("tid", String.valueOf(i))
                        .addField("v1", NumberUtil.round(Math.random(), 3).doublevalue())
                        .addField("v2", NumberUtil.round(Math.random(), 3).doublevalue())
                        .addField("v3", NumberUtil.round(Math.random(), 3).doublevalue());
                records.add(pointBuilder.build().lineProtocol());
            }
        }
        return records;
    }

    
    private static Connection getTdConn() throws Exception {
        Class.forName("com.taosdata.jdbc.TSDBDriver");
        // Class.forName("com.taosdata.jdbc.rs.RestfulDriver");
        String jdbcUrl = "jdbc:TAOS://elephant:6030/db_wlf?user=root&password=taosdata";
        // String jdbcUrl = "jdbc:TAOS-RS://elephant:6041/db_wlf?user=root&password=taosdata";
        Properties connProps = new Properties();
        connProps.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
        connProps.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
        connProps.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
        return DriverManager.getConnection(jdbcUrl, connProps);
    }

    
    private static InfluxDB getInfluxDB() {
        String serverURL = "http://elephant:8086";
        String username = "wlf";
        String password = "wlf@123";
        String database = "db_wlf";
        OkHttpClient.Builder clientBuilder = new OkHttpClient.Builder().readTimeout(100, TimeUnit.SECONDS);
        // 批大小（万条）
        int batchSize = 20;
        // 写间隔（ms）
        int interval = 1000;
        return InfluxDBFactory
                .connect(serverURL, username, password, clientBuilder)
                .setDatabase(database)
                .enableGzip();
        // 第一个是point的个数，第二个是时间，单位毫秒，第三个时间单位
//                .enableBatch(batchSize * 10000, interval, TimeUnit.MILLISECONDS);
    }
}

TDengineGUI这个工具界面看着挺素的，但是使用起来还是挺方便的。不过没有提供数据的导入和导出功能（官方命令行提供数据的导入和导出）。