Skip to main content

TDengine Java Connector

taos-jdbcdriver 是 TDengine 的官方 Java 语言连接器，Java 开发人员可以通过它开发存取 TDengine 数据库的应用软件。taos-jdbcdriver 实现了 JDBC driver 标准的接口.

info

TDengine 的 JDBC 驱动实现尽可能与关系型数据库驱动保持一致，但 TDengine 与关系对象型数据库的使用场景和技术特征存在差异，所以taos-jdbcdriver 与传统的 JDBC driver 也存在一定差异。在使用时需要注意以下几点：

  • TDengine 目前不支持针对单条数据记录的删除操作。
  • 目前不支持事务操作。

连接方式​

taos-jdbcdriver主要提供三种形式的连接器。一般我们推荐使用 Websocket 连接。

  • 原生连接，通过 TDengine 客户端驱动程序（taosc）原生连接 TDengine 实例，支持数据写入、查询、数据订阅、schemaless 接口和参数绑定接口等功能。
  • REST 连接，通过 taosAdapter 提供的 HTTP 接口连接 TDengine 实例，不支持 schemaless 和数据订阅等特性。
  • Websocket 连接，通过 taosAdapter 提供的 Websocket 接口连接 TDengine 实例，WebSocket 连接实现的功能集合和原生连接有少量不同。

连接方式的详细介绍请参考：连接器建立连接的方式

JDBC 和 JRE 兼容性​

  • JDBC: 主要兼容 JDBC 4.2 版本，部分功能如无模式写入和数据订阅单独提供
  • JRE: 支持 JRE 8 及以上版本

支持的平台​

原生连接支持的平台和 TDengine 客户端驱动支持的平台一致。 REST 连接支持所有能运行 Java 的平台。

版本历史​

taos-jdbcdriver 版本主要变化TDengine 版本
3.2.9解决了 Websocket prepareStatement 关闭 bug-
3.2.8优化了自动提交, 解决了 websocket 手动提交 bug, 优化 Websocket prepareStatement 使用一个连接, 元数据支持视图-
3.2.7支持 VARBINARY 和 GEOMETRY 类型，增加 native 连接的时区设置支持。增加 websocket 自动重连功能。3.2.0.0 及更高版本
3.2.5数据订阅增加 committed()、assignment() 方法3.1.0.3 及更高版本
3.2.4数据订阅在 WebSocket 连接下增加 enable.auto.commit 参数，以及 unsubscribe() 方法。-
3.2.3修复 ResultSet 在一些情况数据解析失败-
3.2.2新增功能：数据订阅支持 seek 功能。3.0.5.0 及更高版本
3.2.1新增功能：WebSocket 连接支持 schemaless 与 prepareStatement 写入。变更：consumer poll 返回结果集为 ConsumerRecord，可通过 value() 获取指定结果集数据。3.0.3.0 及更高版本
3.2.0存在连接问题，不推荐使用-
3.1.0WebSocket 连接支持订阅功能-
3.0.1 - 3.0.4修复一些情况下结果集数据解析错误的问题。3.0.1 在 JDK 11 环境编译，JDK 8 环境下建议使用其他版本-
3.0.0支持 TDengine 3.03.0.0.0 及更高版本
2.0.42修复 WebSocket 连接中 wasNull 接口返回值-
2.0.41修复 REST 连接中用户名和密码转码方式-
2.0.39 - 2.0.40增加 REST 连接/请求 超时设置-
2.0.38JDBC REST 连接增加批量拉取功能-
2.0.37增加对 json tag 支持-
2.0.36增加对 schemaless 写入支持-

处理异常​

在报错后，通过 SQLException 可以获取到错误的信息和错误码：

try (Statement statement = connection.createStatement()) {
    // executeQuery
    ResultSet tempResultSet = statement.executeQuery(sql);
    // print result
    printResult(tempResultSet);
} catch (SQLException e) {
    System.out.println("ERROR Message: " + e.getMessage());
    System.out.println("ERROR Code: " + e.getErrorCode());
    e.printStackTrace();
}

查看源码

JDBC 连接器可能报错的错误码包括 4 种：

  • JDBC driver 本身的报错（错误码在 0x2301 到 0x2350 之间）
  • 原生连接方法的报错（错误码在 0x2351 到 0x2360 之间）
  • 数据订阅的报错（错误码在 0x2371 到 0x2380 之间）
  • TDengine 其他功能模块的报错。

具体的错误码请参考：

Error CodeDescriptionSuggested Actions
0x2301connection already closed连接已经关闭，检查连接情况，或重新创建连接去执行相关指令。
0x2302this operation is NOT supported currently!当前使用接口不支持，可以更换其他连接方式。
0x2303invalid variables参数不合法，请检查相应接口规范，调整参数类型及大小。
0x2304statement is closedstatement 已经关闭，请检查 statement 是否关闭后再次使用，或是连接是否正常。
0x2305resultSet is closedresultSet 结果集已经释放，请检查 resultSet 是否释放后再次使用。
0x2306Batch is empty!prepareStatement 添加参数后再执行 executeBatch。
0x2307Can not issue data manipulation statements with executeQuery()更新操作应该使用 executeUpdate()，而不是 executeQuery()。
0x2308Can not issue SELECT via executeUpdate()查询操作应该使用 executeQuery()，而不是 executeUpdate()。
0x230dparameter index out of range参数越界，请检查参数的合理范围。
0x230econnection already closed连接已经关闭，请检查 Connection 是否关闭后再次使用，或是连接是否正常。
0x230funknown sql type in tdengine请检查 TDengine 支持的 Data Type 类型。
0x2310can't register JDBC-JNI driver不能注册 JNI 驱动，请检查 url 是否填写正确。
0x2312url is not set请检查 REST 连接 url 是否填写正确。
0x2314numeric value out of range请检查获取结果集中数值类型是否使用了正确的接口。
0x2315unknown taos type in tdengine在 TDengine 数据类型与 JDBC 数据类型转换时，是否指定了正确的 TDengine 数据类型。
0x2317REST 连接中使用了错误的请求类型。
0x2318REST 连接中出现了数据传输异常，请检查网络情况并重试。
0x2319user is required创建连接时缺少用户名信息
0x231apassword is required创建连接时缺少密码信息
0x231chttpEntity is null, sql:REST 连接中执行出现异常
0x231dcan't create connection with server within通过增加参数 httpConnectTimeout 增加连接耗时，或是请检查与 taosAdapter 之间的连接情况。
0x231efailed to complete the task within the specified time通过增加参数 messageWaitTimeout 增加执行耗时，或是请检查与 taosAdapter 之间的连接情况。
0x2350unknown error未知异常，请在 github 反馈给开发人员。
0x2352Unsupported encoding本地连接下指定了不支持的字符编码集
0x2353internal error of database, please see taoslog for more details本地连接执行 prepareStatement 时出现错误，请检查 taos log 进行问题定位。
0x2354JNI connection is NULL本地连接执行命令时，Connection 已经关闭。请检查与 TDengine 的连接情况。
0x2355JNI result set is NULL本地连接获取结果集，结果集异常，请检查连接情况，并重试。
0x2356invalid num of fields本地连接获取结果集的 meta 信息不匹配。
0x2357empty sql string填写正确的 SQL 进行执行。
0x2359JNI alloc memory failed, please see taoslog for more details本地连接分配内存错误，请检查 taos log 进行问题定位。
0x2371consumer properties must not be null!创建订阅时参数为空，请填写正确的参数。
0x2372configs contain empty key, failed to set consumer property参数 key 中包含空值，请填写正确的参数。
0x2373failed to set consumer property,参数 value 中包含空值，请填写正确的参数。
0x2375topic reference has been destroyed创建数据订阅过程中，topic 引用被释放。请检查与 TDengine 的连接情况。
0x2376failed to set consumer topic, topic name is empty创建数据订阅过程中，订阅 topic 名称为空。请检查指定的 topic 名称是否填写正确。
0x2377consumer reference has been destroyed订阅数据传输通道已经关闭，请检查与 TDengine 的连接情况。
0x2378consumer create error创建数据订阅失败，请根据错误信息检查 taos log 进行问题定位。
0x2379seek offset must not be a negative numberseek 接口参数不能为负值，请使用正确的参数
0x237avGroup not found in result setVGroup 没有分配给当前 consumer，由于 Rebalance 机制导致 Consumer 与 VGroup 不是绑定的关系

TDengine DataType 和 Java DataType​

TDengine 目前支持时间戳、数字、字符、布尔类型，与 Java 对应类型转换如下：

TDengine DataTypeJDBCType
TIMESTAMPjava.sql.Timestamp
INTjava.lang.Integer
BIGINTjava.lang.Long
FLOATjava.lang.Float
DOUBLEjava.lang.Double
SMALLINTjava.lang.Short
TINYINTjava.lang.Byte
BOOLjava.lang.Boolean
BINARYbyte array
NCHARjava.lang.String
JSONjava.lang.String
VARBINARYbyte[]
GEOMETRYbyte[]

注意：JSON 类型仅在 tag 中支持。
由于历史原因，TDengine中的BINARY底层不是真正的二进制数据，已不建议使用。请用VARBINARY类型代替。
GEOMETRY类型是little endian字节序的二进制数据，符合WKB规范。详细信息请参考 数据类型
WKB规范请参考Well-Known Binary (WKB)
对于java连接器，可以使用jts库来方便的创建GEOMETRY类型对象，序列化后写入TDengine，这里有一个样例Geometry示例

安装步骤​

安装前准备​

使用 Java Connector 连接数据库前，需要具备以下条件：

  • 已安装 Java 1.8 或以上版本运行时环境和 Maven 3.6 或以上版本
  • 已安装 TDengine 客户端驱动（使用原生连接必须安装，使用 REST 连接无需安装），具体步骤请参考安装客户端驱动

安装连接器​

目前 taos-jdbcdriver 已经发布到 Sonatype Repository 仓库，且各大仓库都已同步。

Maven 项目中，在 pom.xml 中添加以下依赖：

<dependency>
  <groupId>com.taosdata.jdbc</groupId>
  <artifactId>taos-jdbcdriver</artifactId>
  <version>3.2.9</version>
</dependency>

建立连接​

TDengine 的 JDBC URL 规范格式为： jdbc:[TAOS|TAOS-RS]://[host_name]:[port]/[database_name]?[user={user}|&password={password}|&charset={charset}|&cfgdir={config_dir}|&locale={locale}|&timezone={timezone}]

对于建立连接，原生连接与 REST 连接有细微不同。

注：REST 连接中增加 batchfetch 参数并设置为 true，将开启 WebSocket 连接。

Class.forName("com.taosdata.jdbc.rs.RestfulDriver");
String jdbcUrl = "jdbc:TAOS-RS://taosdemo.com:6041/power?user=root&password=taosdata";
Connection conn = DriverManager.getConnection(jdbcUrl);

以上示例，使用了 JDBC REST 连接的 RestfulDriver，建立了到 hostname 为 taosdemo.com，端口为 6041，数据库名为 power 的连接。这个 URL 中指定用户名（user）为 root，密码（password）为 taosdata。

使用 JDBC REST 连接，不需要依赖客户端驱动。与 JDBC 原生连接相比，仅需要：

  1. driverClass 指定为“com.taosdata.jdbc.rs.RestfulDriver”；
  2. jdbcUrl 以“jdbc:TAOS-RS://”开头；
  3. 使用 6041 作为连接端口。

url 中的配置参数如下：

  • user：登录 TDengine 用户名，默认值 'root'。
  • password：用户登录密码，默认值 'taosdata'。
  • batchfetch: true：在执行查询时批量拉取结果集；false：逐行拉取结果集。默认值为：false。逐行拉取结果集使用 HTTP 方式进行数据传输。JDBC REST 连接支持批量拉取数据功能。taos-jdbcdriver 与 TDengine 之间通过 WebSocket 连接进行数据传输。相较于 HTTP，WebSocket 可以使 JDBC REST 连接支持大数据量查询，并提升查询性能。
  • charset: 当开启批量拉取数据时，指定解析字符串数据的字符集。
  • batchErrorIgnore：true：在执行 Statement 的 executeBatch 时，如果中间有一条 SQL 执行失败，继续执行下面的 SQL 了。false：不再执行失败 SQL 后的任何语句。默认值为：false。
  • httpConnectTimeout: 连接超时时间，单位 ms， 默认值为 60000。
  • httpSocketTimeout: socket 超时时间，单位 ms，默认值为 60000。仅在 batchfetch 设置为 false 时生效。
  • messageWaitTimeout: 消息超时时间, 单位 ms， 默认值为 60000。 仅在 batchfetch 设置为 true 时生效。
  • useSSL: 连接中是否使用 SSL。
  • httpPoolSize: REST 并发请求大小，默认 20。

注意：部分配置项（比如：locale、timezone）在 REST 连接中不生效。

note
  • 与原生连接方式不同，REST 接口是无状态的。在使用 JDBC REST 连接时，需要在 SQL 中指定表、超级表的数据库名称。例如：
INSERT INTO power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') VALUES (NOW, 10.30000, 219, 0.31000);
  • 如果在 url 中指定了 dbname，那么，JDBC REST 连接会默认使用/rest/sql/dbname 作为 restful 请求的 url，在 SQL 中不需要指定 dbname。例如：url 为 jdbc:TAOS-RS://127.0.0.1:6041/power，那么，可以执行 sql：INSERT INTO d1001 USING meters TAGS(2,'California.SanFrancisco') VALUES (NOW, 10.30000, 219, 0.31000);

指定 URL 和 Properties 获取连接​

除了通过指定的 URL 获取连接，还可以使用 Properties 指定建立连接时的参数。

注意：

  • 应用中设置的 client parameter 为进程级别的，即如果要更新 client 的参数，需要重启应用。这是因为 client parameter 是全局参数，仅在应用程序的第一次设置生效。
  • 以下示例代码基于 taos-jdbcdriver-3.1.0 或以上版本。
public Connection getConn() throws Exception{
  Class.forName("com.taosdata.jdbc.TSDBDriver");
  String jdbcUrl = "jdbc:TAOS://taosdemo.com:6030/power?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");
  connProps.setProperty("debugFlag", "135");
  Connection conn = DriverManager.getConnection(jdbcUrl, connProps);
  return conn;
}

public Connection getRestConn() throws Exception{
  Class.forName("com.taosdata.jdbc.rs.RestfulDriver");
  String jdbcUrl = "jdbc:TAOS-RS://taosdemo.com:6041/power?user=root&password=taosdata";
  Properties connProps = new Properties();
  connProps.setProperty(TSDBDriver.PROPERTY_KEY_BATCH_LOAD, "true");
  Connection conn = DriverManager.getConnection(jdbcUrl, connProps);
  return conn;
}

以上示例，建立一个到 hostname 为 taosdemo.com，端口为 6030/6041，数据库名为 power 的连接。这个连接在 url 中指定了用户名(user)为 root，密码（password）为 taosdata，并在 connProps 中指定了使用的字符集、语言环境、时区、是否开启批量拉取等信息。

properties 中的配置参数如下：

  • TSDBDriver.PROPERTY_KEY_USER：登录 TDengine 用户名，默认值 'root'。
  • TSDBDriver.PROPERTY_KEY_PASSWORD：用户登录密码，默认值 'taosdata'。
  • TSDBDriver.PROPERTY_KEY_BATCH_LOAD: true：在执行查询时批量拉取结果集；false：逐行拉取结果集。默认值为：false。
  • TSDBDriver.PROPERTY_KEY_BATCH_ERROR_IGNORE：true：在执行 Statement 的 executeBatch 时，如果中间有一条 SQL 执行失败，继续执行下面的 sq 了。false：不再执行失败 SQL 后的任何语句。默认值为：false。
  • TSDBDriver.PROPERTY_KEY_CONFIG_DIR：仅在使用 JDBC 原生连接时生效。客户端配置文件目录路径，Linux OS 上默认值 /etc/taos，Windows OS 上默认值 C:/TDengine/cfg。
  • TSDBDriver.PROPERTY_KEY_CHARSET：客户端使用的字符集，默认值为系统字符集。
  • TSDBDriver.PROPERTY_KEY_LOCALE：仅在使用 JDBC 原生连接时生效。 客户端语言环境，默认值系统当前 locale。
  • TSDBDriver.PROPERTY_KEY_TIME_ZONE：仅在使用 JDBC 原生连接时生效。 客户端使用的时区，默认值为系统当前时区。因为历史的原因，我们只支持POSIX标准的部分规范，如UTC-8(代表中国上上海), GMT-8，Asia/Shanghai 这几种形式。
  • TSDBDriver.HTTP_CONNECT_TIMEOUT: 连接超时时间，单位 ms， 默认值为 60000。仅在 REST 连接时生效。
  • TSDBDriver.HTTP_SOCKET_TIMEOUT: socket 超时时间，单位 ms，默认值为 60000。仅在 REST 连接且 batchfetch 设置为 false 时生效。
  • TSDBDriver.PROPERTY_KEY_MESSAGE_WAIT_TIMEOUT: 消息超时时间, 单位 ms， 默认值为 60000。 仅在 REST 连接且 batchfetch 设置为 true 时生效。
  • TSDBDriver.PROPERTY_KEY_USE_SSL: 连接中是否使用 SSL。仅在 REST 连接时生效。
  • TSDBDriver.HTTP_POOL_SIZE: REST 并发请求大小，默认 20。

此外对 JDBC 原生连接，通过指定 URL 和 Properties 还可以指定其他参数，比如日志级别、SQL 长度等。更多详细配置请参考客户端配置。

配置参数的优先级​

通过前面三种方式获取连接，如果配置参数在 url、Properties、客户端配置文件中有重复，则参数的优先级由高到低分别如下：

  1. JDBC URL 参数，如上所述，可以在 JDBC URL 的参数中指定。
  2. Properties connProps
  3. 使用原生连接时，TDengine 客户端驱动的配置文件 taos.cfg

例如：在 url 中指定了 password 为 taosdata，在 Properties 中指定了 password 为 taosdemo，那么，JDBC 会使用 url 中的 password 建立连接。

使用示例​

创建数据库和表​

// create database
stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS power");

// use database
stmt.executeUpdate("USE power");

// create table
stmt.executeUpdate("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))");

查看源码

注意：如果不使用 USE power 指定数据库，则后续对表的操作都需要增加数据库名称作为前缀，如 power.meters。

插入数据​

// insert data
String insertQuery = "INSERT INTO " +
    "power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " +
    "VALUES " +
    "(NOW + 1a, 10.30000, 219, 0.31000) " +
    "(NOW + 2a, 12.60000, 218, 0.33000) " +
    "(NOW + 3a, 12.30000, 221, 0.31000) " +
    "power.d1002 USING power.meters TAGS(3, 'California.SanFrancisco') " +
    "VALUES " +
    "(NOW + 1a, 10.30000, 218, 0.25000) ";
int affectedRows = stmt.executeUpdate(insertQuery);
System.out.println("insert " + affectedRows + " rows.");

查看源码

NOW 为系统内部函数，默认为客户端所在计算机当前时间。 NOW + 1s 代表客户端当前时间往后加 1 秒，数字后面代表时间单位：a(毫秒)，s(秒)，m(分)，h(小时)，d(天)，w(周)，n(月)，y(年)。

查询数据​

// query data
ResultSet resultSet = stmt.executeQuery("SELECT * FROM meters");

Timestamp ts;
float current;
String location;
while(resultSet.next()){
    ts = resultSet.getTimestamp(1);
    current = resultSet.getFloat(2);
    location = resultSet.getString("location");

    System.out.printf("%s, %f, %s\n", ts, current, location);
}

查看源码

查询和操作关系型数据库一致，使用下标获取返回字段内容时从 1 开始，建议使用字段名称获取。

执行带有 reqId 的 SQL​

reqId 可用于请求链路追踪，reqId 就像分布式系统中的 traceId 作用一样。一个请求可能需要经过多个服务或者模块才能完成。reqId 用于标识和关联这个请求的所有相关操作，以便于我们可以追踪和分析请求的完整路径。
使用 reqId 有下面好处：

  • 请求追踪：通过将同一个 reqId 关联到一个请求的所有相关操作，可以追踪请求在系统中的完整路径
  • 性能分析：通过分析一个请求的 reqId，可以了解请求在各个服务和模块中的处理时间，从而找出性能瓶颈
  • 故障诊断：当一个请求失败时，可以通过查看与该请求关联的 reqId 来找出问题发生的位置

如果用户不设置reqId，连接器也会内部随机生成一个，但是还是建议用户设置，可以更好的跟用户请求关联起来。

AbstractStatement aStmt = (AbstractStatement) connection.createStatement();
aStmt.execute("CREATE DATABASE IF NOT EXISTS power", 1L);
aStmt.executeUpdate("USE power", 2L);
try (ResultSet rs = aStmt.executeQuery("SELECT * FROM meters limit 1", 3L)) {
    while(rs.next()){
        Timestamp timestamp = rs.getTimestamp(1);
        System.out.println("timestamp = " + timestamp);
    }
}
aStmt.close();

查看源码

通过参数绑定写入数据​

TDengine 的 JDBC 原生连接实现大幅改进了参数绑定方式对数据写入（INSERT）场景的支持。采用这种方式写入数据时，能避免 SQL 语法解析的资源消耗，从而在很多情况下显著提升写入性能。

注意：

  • JDBC REST 连接目前不支持参数绑定
  • 以下示例代码基于 taos-jdbcdriver-3.2.1 或以上版本
  • binary 类型数据需要调用 setString 方法，nchar 类型数据需要调用 setNString 方法
  • 预处理语句中指定数据库与子表名称不要使用 db.?，应直接使用 ?，然后在 setTableName 中指定数据库，如：prepareStatement.setTableName("db.t1")。
public class ParameterBindingBasicDemo {

    // modify host to your own
    private static final String host = "127.0.0.1";
    private static final Random random = new Random(System.currentTimeMillis());
    private static final int numOfSubTable = 10, numOfRow = 10;

    public static void main(String[] args) throws SQLException {

        String jdbcUrl = "jdbc:TAOS://" + host + ":6030/";
        Connection conn = DriverManager.getConnection(jdbcUrl, "root", "taosdata");

        init(conn);

        String sql = "INSERT INTO ? USING meters TAGS(?,?) VALUES (?,?,?,?)";

        try (TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class)) {

            for (int i = 1; i <= numOfSubTable; i++) {
                // set table name
                pstmt.setTableName("d_bind_" + i);

                // set tags
                pstmt.setTagInt(0, i);
                pstmt.setTagString(1, "location_" + i);

                // set column ts
                ArrayList<Long> tsList = new ArrayList<>();
                long current = System.currentTimeMillis();
                for (int j = 0; j < numOfRow; j++)
                    tsList.add(current + j);
                pstmt.setTimestamp(0, tsList);

                // set column current
                ArrayList<Float> f1List = new ArrayList<>();
                for (int j = 0; j < numOfRow; j++)
                    f1List.add(random.nextFloat() * 30);
                pstmt.setFloat(1, f1List);

                // set column voltage
                ArrayList<Integer> f2List = new ArrayList<>();
                for (int j = 0; j < numOfRow; j++)
                    f2List.add(random.nextInt(300));
                pstmt.setInt(2, f2List);

                // set column phase
                ArrayList<Float> f3List = new ArrayList<>();
                for (int j = 0; j < numOfRow; j++)
                    f3List.add(random.nextFloat());
                pstmt.setFloat(3, f3List);
                // add column
                pstmt.columnDataAddBatch();
            }
            // execute column
            pstmt.columnDataExecuteBatch();
        }

        conn.close();
    }

    private static void init(Connection conn) throws SQLException {
        try (Statement stmt = conn.createStatement()) {
            stmt.execute("CREATE DATABASE IF NOT EXISTS power");
            stmt.execute("USE power");
            stmt.execute("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))");
        }
    }
}

查看源码

这是一个更详细的参数绑定示例

用于设定 VALUES 数据列的取值的方法总共有：

public void setInt(int columnIndex, ArrayList<Integer> list) throws SQLException
public void setFloat(int columnIndex, ArrayList<Float> list) throws SQLException
public void setTimestamp(int columnIndex, ArrayList<Long> list) throws SQLException
public void setLong(int columnIndex, ArrayList<Long> list) throws SQLException
public void setDouble(int columnIndex, ArrayList<Double> list) throws SQLException
public void setBoolean(int columnIndex, ArrayList<Boolean> list) throws SQLException
public void setByte(int columnIndex, ArrayList<Byte> list) throws SQLException
public void setShort(int columnIndex, ArrayList<Short> list) throws SQLException
public void setString(int columnIndex, ArrayList<String> list, int size) throws SQLException
public void setNString(int columnIndex, ArrayList<String> list, int size) throws SQLException
public void setVarbinary(int columnIndex, ArrayList<byte[]> list, int size) throws SQLException
public void setGeometry(int columnIndex, ArrayList<byte[]> list, int size) throws SQLException

注：字符串和数组类型都要求用户在 size 参数里声明表定义中对应列的列宽。

用于设定 TAGS 取值的方法总共有：

public void setTagNull(int index, int type)
public void setTagBoolean(int index, boolean value)
public void setTagInt(int index, int value)
public void setTagByte(int index, byte value)
public void setTagShort(int index, short value)
public void setTagLong(int index, long value)
public void setTagTimestamp(int index, long value)
public void setTagFloat(int index, float value)
public void setTagDouble(int index, double value)
public void setTagString(int index, String value)
public void setTagNString(int index, String value)
public void setTagJson(int index, String value)
public void setTagVarbinary(int index, byte[] value)
public void setTagGeometry(int index, byte[] value)

无模式写入​

TDengine 支持无模式写入功能。无模式写入兼容 InfluxDB 的 行协议（Line Protocol）、OpenTSDB 的 telnet 行协议和 OpenTSDB 的 JSON 格式协议。详情请参见无模式写入。

public class SchemalessJniTest {
    private static final String host = "127.0.0.1";
    private static final String lineDemo = "meters,groupid=2,location=California.SanFrancisco current=10.3000002f64,voltage=219i32,phase=0.31f64 1626006833639000000";
    private static final String telnetDemo = "stb0_0 1707095283260 4 host=host0 interface=eth0";
    private static final String jsonDemo = "{\"metric\": \"meter_current\",\"timestamp\": 1626846400,\"value\": 10.3, \"tags\": {\"groupid\": 2, \"location\": \"California.SanFrancisco\", \"id\": \"d1001\"}}";

    public static void main(String[] args) throws SQLException {
        final String url = "jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata";
        try (Connection connection = DriverManager.getConnection(url)) {
            init(connection);

            SchemalessWriter writer = new SchemalessWriter(connection);
            writer.write(lineDemo, SchemalessProtocolType.LINE, SchemalessTimestampType.NANO_SECONDS);
            writer.write(telnetDemo, SchemalessProtocolType.TELNET, SchemalessTimestampType.MILLI_SECONDS);
            writer.write(jsonDemo, SchemalessProtocolType.JSON, SchemalessTimestampType.NOT_CONFIGURED);
        }
    }

    private static void init(Connection connection) throws SQLException {
        try (Statement stmt = connection.createStatement()) {
            stmt.execute("CREATE DATABASE IF NOT EXISTS power");
            stmt.execute("USE power");
        }
    }
}

查看源码

执行带有 reqId 的无模式写入​

此 reqId 可用于请求链路追踪。

writer.write(lineDemo, SchemalessProtocolType.LINE, SchemalessTimestampType.NANO_SECONDS, 1L);

数据订阅​

TDengine Java 连接器支持订阅功能，应用 API 如下：

创建 Topic​

Connection connection = DriverManager.getConnection(url, properties);
Statement statement = connection.createStatement();
statement.executeUpdate("CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM meters");

查看源码

subscribe 方法的三个参数含义如下：

  • topic_meters：订阅的主题（即名称），此参数是订阅的唯一标识。
  • sql：订阅的查询语句，此语句只能是 SELECT 语句，只应查询原始数据，只能按时间正序查询数据。

如上面的例子将使用 SQL 语句 SELECT ts, current, voltage, phase, groupid, location FROM meters 创建一个名为 topic_meters 的订阅。

创建 Consumer​

Properties config = new Properties();
config.setProperty("td.connect.type", "jni");
config.setProperty("bootstrap.servers", "localhost:6030");
config.setProperty("auto.offset.reset", "latest");
config.setProperty("msg.with.table.name", "true");
config.setProperty("enable.auto.commit", "true");
config.setProperty("auto.commit.interval.ms", "1000");
config.setProperty("group.id", "group1");
config.setProperty("client.id", "1");
config.setProperty("value.deserializer", "com.taosdata.example.AbsConsumerLoop$ResultDeserializer");
config.setProperty("value.deserializer.encoding", "UTF-8");

this.consumer = new TaosConsumer<>(config);

查看源码

  • td.connect.type: 连接方式。jni：表示使用动态库连接的方式，ws/WebSocket：表示使用 WebSocket 进行数据通信。默认为 jni 方式。
  • bootstrap.servers: TDengine 服务端所在的ip:port，如果使用 WebSocket 连接，则为 taosAdapter 所在的ip:port。
  • enable.auto.commit: 是否允许自动提交。
  • group.id: consumer: 所在的 group。
  • value.deserializer: 结果集反序列化方法，可以继承 com.taosdata.jdbc.tmq.ReferenceDeserializer，并指定结果集 bean，实现反序列化。也可以继承 com.taosdata.jdbc.tmq.Deserializer，根据 SQL 的 resultSet 自定义反序列化方式。
  • httpConnectTimeout: 创建连接超时参数，单位 ms，默认为 5000 ms。仅在 WebSocket 连接下有效。
  • messageWaitTimeout: 数据传输超时参数，单位 ms，默认为 10000 ms。仅在 WebSocket 连接下有效。
  • httpPoolSize: 同一个连接下最大并行请求数。仅在 WebSocket 连接下有效。

其他参数请参考：Consumer 参数列表， 注意TDengine服务端自3.2.0.0版本开始消息订阅中的auto.offset.reset默认值发生变化。

订阅消费数据​

while (!shutdown.get()) {
    ConsumerRecords<ResultBean> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<ResultBean> record : records) {
        ResultBean bean = record.value();
        process(bean);
    }
}

查看源码

poll 每次调用获取一个消息。

指定订阅 Offset​

// 获取订阅的 topicPartition
Set<TopicPartition> assignment() throws SQLException;
// 获取 offset
long position(TopicPartition partition) throws SQLException;
Map<TopicPartition, Long> position(String topic) throws SQLException;
Map<TopicPartition, Long> beginningOffsets(String topic) throws SQLException;
Map<TopicPartition, Long> endOffsets(String topic) throws SQLException;
Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions) throws SQLException;

// 指定下一次 poll 中使用的 offset
void seek(TopicPartition partition, long offset) throws SQLException;
void seekToBeginning(Collection<TopicPartition> partitions) throws SQLException;
void seekToEnd(Collection<TopicPartition> partitions) throws SQLException;

示例代码：

String topic = "topic_meters";
Map<TopicPartition, Long> offset = null;
try (TaosConsumer<AbsConsumerLoop.ResultBean> consumer = new TaosConsumer<>(config)) {
    consumer.subscribe(Collections.singletonList(topic));
    for (int i = 0; i < 10; i++) {
        if (i == 3) {
            // Saving consumption position
            offset = consumer.position(topic);
        }
        if (i == 5) {
            // reset consumption to the previously saved position
            for (Map.Entry<TopicPartition, Long> entry : offset.entrySet()) {
                consumer.seek(entry.getKey(), entry.getValue());
            }
        }
        ConsumerRecords<AbsConsumerLoop.ResultBean> records = consumer.poll(Duration.ofMillis(500));
    }
}

查看源码

提交 Offset​

当enable.auto.commit为 false 时，可以手动提交 offset。

void commitSync() throws SQLException;
void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) throws SQLException;
// 异步提交仅在 native 连接下有效
void commitAsync(OffsetCommitCallback<V> callback) throws SQLException;
void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback<V> callback) throws SQLException;

关闭订阅​

// 取消订阅
consumer.unsubscribe();
// 关闭消费
consumer.close()

详情请参考：数据订阅

完整示例​

public abstract class AbsConsumerLoopFull {
    private final TaosConsumer<ResultBean> consumer;
    private final List<String> topics;
    private final AtomicBoolean shutdown;
    private final CountDownLatch shutdownLatch;

    public AbsConsumerLoopFull() throws SQLException {

        Properties config = new Properties();
        config.setProperty("td.connect.type", "jni");
        config.setProperty("bootstrap.servers", "localhost:6030");
        config.setProperty("auto.offset.reset", "latest");
        config.setProperty("msg.with.table.name", "true");
        config.setProperty("enable.auto.commit", "true");
        config.setProperty("auto.commit.interval.ms", "1000");
        config.setProperty("group.id", "group1");
        config.setProperty("client.id", "1");
        config.setProperty("value.deserializer", "com.taosdata.example.AbsConsumerLoop$ResultDeserializer");
        config.setProperty("value.deserializer.encoding", "UTF-8");

        this.consumer = new TaosConsumer<>(config);

        this.topics = Collections.singletonList("topic_meters");
        this.shutdown = new AtomicBoolean(false);
        this.shutdownLatch = new CountDownLatch(1);
    }

    public abstract void process(ResultBean result);

    public void pollData() throws SQLException {
        try {
            consumer.subscribe(topics);
            while (!shutdown.get()) {
                ConsumerRecords<ResultBean> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<ResultBean> record : records) {
                    ResultBean bean = record.value();
                    process(bean);
                }
            }
            consumer.unsubscribe();
        } finally {
            consumer.close();
            shutdownLatch.countDown();
        }
    }

    public void shutdown() throws InterruptedException {
        shutdown.set(true);
        shutdownLatch.await();
    }

    public static class ResultDeserializer extends ReferenceDeserializer<ResultBean> {

    }

    public static class ResultBean {
        private Timestamp ts;
        private double current;
        private int voltage;
        private double phase;
        private int groupid;
        private String location;

        public Timestamp getTs() {
            return ts;
        }

        public void setTs(Timestamp ts) {
            this.ts = ts;
        }

        public double getCurrent() {
            return current;
        }

        public void setCurrent(double current) {
            this.current = current;
        }

        public int getVoltage() {
            return voltage;
        }

        public void setVoltage(int voltage) {
            this.voltage = voltage;
        }

        public double getPhase() {
            return phase;
        }

        public void setPhase(double phase) {
            this.phase = phase;
        }

        public int getGroupid() {
            return groupid;
        }

        public void setGroupid(int groupid) {
            this.groupid = groupid;
        }

        public String getLocation() {
            return location;
        }

        public void setLocation(String location) {
            this.location = location;
        }
    }
}

查看源码

注意：这里的 value.deserializer 配置参数值应该根据测试环境的包路径做相应的调整。

与连接池使用​

HikariCP​

使用示例如下：

public static void main(String[] args) throws SQLException {
    HikariConfig config = new HikariConfig();
    // jdbc properties
    config.setJdbcUrl("jdbc:TAOS://127.0.0.1:6030/log");
    config.setUsername("root");
    config.setPassword("taosdata");
    // connection pool configurations
    config.setMinimumIdle(10);           //minimum number of idle connection
    config.setMaximumPoolSize(10);      //maximum number of connection in the pool
    config.setConnectionTimeout(30000); //maximum wait milliseconds for get connection from pool
    config.setMaxLifetime(0);       // maximum life time for each connection
    config.setIdleTimeout(0);       // max idle time for recycle idle connection
    config.setConnectionTestQuery("SELECT SERVER_STATUS()"); //validation query

    HikariDataSource ds = new HikariDataSource(config); //create datasource

    Connection  connection = ds.getConnection(); // get connection
    Statement statement = connection.createStatement(); // get statement

    //query or insert
    // ...
    statement.close();
    connection.close(); // put back to connection pool
}

查看源码

通过 HikariDataSource.getConnection() 获取连接后，使用完成后需要调用 close() 方法，实际上它并不会关闭连接，只是放回连接池中。 更多 HikariCP 使用问题请查看官方说明。

Druid​

使用示例如下：

public static void main(String[] args) throws Exception {
    String url = "jdbc:TAOS://127.0.0.1:6030/log";

    DruidDataSource dataSource = new DruidDataSource();
    // jdbc properties
    dataSource.setDriverClassName("com.taosdata.jdbc.TSDBDriver");
    dataSource.setUrl(url);
    dataSource.setUsername("root");
    dataSource.setPassword("taosdata");
    // pool configurations
    dataSource.setInitialSize(10);
    dataSource.setMinIdle(10);
    dataSource.setMaxActive(10);
    dataSource.setMaxWait(30000);
    dataSource.setValidationQuery("SELECT SERVER_STATUS()");

    Connection connection = dataSource.getConnection(); // get connection
    Statement statement = connection.createStatement(); // get statement
    //query or insert
    // ...

    statement.close();
    connection.close(); // put back to connection pool
}

查看源码

更多 druid 使用问题请查看官方说明。

更多示例程序​

示例程序源码位于 TDengine/examples/JDBC 下:

  • JDBCDemo：JDBC 示例源程序。
  • connectionPools：HikariCP, Druid, dbcp, c3p0 等连接池中使用 taos-jdbcdriver。
  • SpringJdbcTemplate：Spring JdbcTemplate 中使用 taos-jdbcdriver。
  • mybatisplus-demo：Springboot + Mybatis 中使用 taos-jdbcdriver。
  • springbootdemo: Springboot 中使用 taos-jdbcdriver。
  • consumer-demo：Consumer 消费 TDengine 数据示例，可通过参数控制消费速度。

请参考：JDBC example

常见问题​

  1. 使用 Statement 的 addBatch() 和 executeBatch() 来执行“批量写入/更新”，为什么没有带来性能上的提升？

原因：TDengine 的 JDBC 实现中，通过 addBatch 方法提交的 SQL 语句，会按照添加的顺序，依次执行，这种方式没有减少与服务端的交互次数，不会带来性能上的提升。

解决方法：1. 在一条 insert 语句中拼接多个 values 值；2. 使用多线程的方式并发插入；3. 使用参数绑定的写入方式

  1. java.lang.UnsatisfiedLinkError: no taos in java.library.path

原因：程序没有找到依赖的本地函数库 taos。

解决方法：Windows 下可以将 C:\TDengine\driver\taos.dll 拷贝到 C:\Windows\System32\ 目录下，Linux 下将建立如下软链 ln -s /usr/local/taos/driver/libtaos.so.x.x.x.x /usr/lib/libtaos.so 即可，macOS 下需要建立软链 ln -s /usr/local/lib/libtaos.dylib。

  1. java.lang.UnsatisfiedLinkError: taos.dll Can't load AMD 64 bit on a IA 32-bit platform

原因：目前 TDengine 只支持 64 位 JDK。

解决方法：重新安装 64 位 JDK。

  1. java.lang.NoSuchMethodError: setByteArray

原因：taos-jdbcdriver 3.* 版本仅支持 TDengine 3.0 及以上版本。

解决方法： 使用 taos-jdbcdriver 2.* 版本连接 TDengine 2.* 版本。

  1. java.lang.NoSuchMethodError: java.nio.ByteBuffer.position(I)Ljava/nio/ByteBuffer; ... taos-jdbcdriver-3.0.1.jar

原因：taos-jdbcdriver 3.0.1 版本需要在 JDK 11+ 环境使用。

解决方法： 更换 taos-jdbcdriver 3.0.2+ 版本。

其它问题请参考 FAQ

API 参考​

taos-jdbcdriver doc

TDengine Database
TDengine 核心是一款高性能、集群开源、云原生的时序数据库（Time Series Database，TSDB），专为物联网、工业互联网、电力、IT 运维等场景设计并优化，具有极强的弹性伸缩能力。同时它还带有内建的缓存、流式计算、数据订阅等系统功能，能大幅减少系统设计的复杂度，降低研发和运营成本，是一个高性能、分布式的物联网、工业大数据平台。
产品
TDengine Enterprise
TDengine Cloud
学习
文档
博客
资源
公司
关于我们
技术支持
新闻动态
联系我们
招贤纳士
服务条款
隐私
场景
物联网
工业互联网
车联网
IT 运维
电力能源
金融
TDengine Database
TDengine Database
加小 T 为好友
即可加入物联网大数据技术前沿群
点击填写表单
获得解决方案专家帮助
© 2023 涛思数据
京公网安备 11010502047618号
新版时序数据库 TDengine v3.0