简体中文 | English
简体中文
taos-jdbcdriver
是 TDengine 的官方 Java 语言连接器,Java 开发人员可以通过它开发存取 TDengine 数据库的应用软件。taos-jdbcdriver
实现了 JDBC driver 标准的接口,并提供两种形式的连接器。一种是通过 TDengine 客户端驱动程序(taosc)原生连接 TDengine 实例,支持数据写入、查询、订阅、schemaless 接口和参数绑定接口等功能,一种是通过 taosAdapter 提供的 REST 接口连接 TDengine 实例(2.0.18 及更高版本)。REST 连接实现的功能集合和原生连接有少量不同。
上图显示了两种 Java 应用使用连接器访问 TDengine 的两种方式:
使用 REST 连接,不依赖 TDengine 客户端驱动,可以跨平台,更加方便灵活,但性能比原生连接器低约 30%。
注意:
TDengine 的 JDBC 驱动实现尽可能与关系型数据库驱动保持一致,但 TDengine 与关系对象型数据库的使用场景和技术特征存在差异,所以taos-jdbcdriver
与传统的 JDBC driver 也存在一定差异。在使用时需要注意以下几点:
原生连接支持的平台和 TDengine 客户端驱动支持的平台一致。
REST 连接支持所有能运行 Java 的平台。
请参考版本支持列表。
TDengine 目前支持时间戳、数字、字符、布尔类型,与 Java 对应类型转换如下:
TDengine DataType | JDBCType (driver 版本 <2.0.24) | JDBCType (driver 版本 >= 2.0.24) |
---|---|---|
TIMESTAMP | java.lang.Long | java.sql.Timestamp |
INT | java.lang.Integer | java.lang.Integer |
BIGINT | java.lang.Long | java.lang.Long |
FLOAT | java.lang.Float | java.lang.Float |
DOUBLE | java.lang.Double | java.lang.Double |
SMALLINT | java.lang.Short | java.lang.Short |
TINYINT | java.lang.Byte | java.lang.Byte |
BOOL | java.lang.Boolean | java.lang.Boolean |
BINARY | java.lang.String | byte array |
NCHAR | java.lang.String | java.lang.String |
JSON | - | java.lang.String |
注意:JSON 类型仅在 tag 中支持。
使用 Java Connector 连接数据库前,需要具备以下条件:
目前 taos-jdbcdriver 已经发布到 Sonatype Repository 仓库,且各大仓库都已同步。
Maven 项目中,在 pom.xml 中添加以下依赖:
taos-jdbcdriver
可以通过下载 TDengine 的源码,自己编译最新版本的 Java connector
git clone https://github.com/taosdata/taos-connector-jdbc.git
cd taos-connector-jdbc
mvn clean install -D maven.test.skip=true
编译后,在 target 目录下会产生 taos-jdbcdriver-2.0.XX-dist.jar 的 jar 包,并自动将编译的 jar 文件放在本地的 Maven 仓库中。
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 连接有细微不同。
Class.forName("com.taosdata.jdbc.TSDBDriver");
String jdbcUrl = "jdbc:TAOS://taosdemo.com:6030/test?user=root&password=taosdata";
Connection cOnn= DriverManager.getConnection(jdbcUrl);
以上示例,使用了 JDBC 原生连接的 TSDBDriver,建立了到 hostname 为 taosdemo.com,端口为 6030(TDengine 的默认端口),数据库名为 test 的连接。这个 URL 中指定用户名(user)为 root,密码(password)为 taosdata。
注意:使用 JDBC 原生连接,taos-jdbcdriver 需要依赖客户端驱动(Linux 下是 libtaos.so;Windows 下是 taos.dll)。
url 中的配置参数如下:
/etc/taos
,Windows OS 上默认值 C:/TDengine/cfg
。JDBC 原生连接的使用请参见视频教程。
当使用 JDBC 原生连接连接 TDengine 集群时,可以使用 TDengine 客户端驱动配置文件,在配置文件中指定集群的 firstEp、secondEp 等参数。如下所示:
public Connection getConn() throws Exception{
Class.forName("com.taosdata.jdbc.TSDBDriver");
String jdbcUrl = "jdbc:TAOS://:/test?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");
Connection cOnn= DriverManager.getConnection(jdbcUrl, connProps);
return conn;
}
# first fully qualified domain name (FQDN) for TDengine system
firstEp cluster_node1:6030
# second fully qualified domain name (FQDN) for TDengine system, for cluster only
secondEp cluster_node2:6030
# default system charset
# charset UTF-8
# system locale
# locale en_US.UTF-8
以上示例,jdbc 会使用客户端的配置文件,建立到 hostname 为 cluster_node1、端口为 6030、数据库名为 test 的连接。当集群中 firstEp 节点失效时,JDBC 会尝试使用 secondEp 连接集群。
TDengine 中,只要保证 firstEp 和 secondEp 中一个节点有效,就可以正常建立到集群的连接。
注意:这里的配置文件指的是调用 JDBC Connector 的应用程序所在机器上的配置文件,Linux OS 上默认值 /etc/taos/taos.cfg ,Windows OS 上默认值 C://TDengine/cfg/taos.cfg。
Class.forName("com.taosdata.jdbc.rs.RestfulDriver");
String jdbcUrl = "jdbc:TAOS-RS://taosdemo.com:6041/test?user=root&password=taosdata";
Connection cOnn= DriverManager.getConnection(jdbcUrl);
以上示例,使用了 JDBC REST 连接的 RestfulDriver,建立了到 hostname 为 taosdemo.com,端口为 6041,数据库名为 test 的连接。这个 URL 中指定用户名(user)为 root,密码(password)为 taosdata。
使用 JDBC REST 连接,不需要依赖客户端驱动。与 JDBC 原生连接相比,仅需要:
url 中的配置参数如下:
注意:部分配置项(比如:locale、timezone)在 REST 连接中不生效。
INSERT INTO test.t1 USING test.weather (ts, temperature) TAGS('beijing') VALUES(now, 24.6);
除了通过指定的 URL 获取连接,还可以使用 Properties 指定建立连接时的参数。
注意:
public Connection getConn() throws Exception{
Class.forName("com.taosdata.jdbc.TSDBDriver");
String jdbcUrl = "jdbc:TAOS://taosdemo.com:6030/test?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");
connProps.setProperty("maxSQLLength", "1048576");
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/test?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,数据库名为 test 的连接。这个连接在 url 中指定了用户名(user)为 root,密码(password)为 taosdata,并在 connProps 中指定了使用的字符集、语言环境、时区、是否开启批量拉取等信息。
properties 中的配置参数如下:
/etc/taos
,Windows OS 上默认值 C:/TDengine/cfg
。通过前面三种方式获取连接,如果配置参数在 url、Properties、客户端配置文件中有重复,则参数的优先级由高到低
分别如下:
例如:在 url 中指定了 password 为 taosdata,在 Properties 中指定了 password 为 taosdemo,那么,JDBC 会使用 url 中的 password 建立连接。
Statement stmt = conn.createStatement();
// create database
stmt.executeUpdate("create database if not exists db");
// use database
stmt.executeUpdate("use db");
// create table
stmt.executeUpdate("create table if not exists tb (ts timestamp, temperature int, humidity float)");
注意:如果不使用
use db
指定数据库,则后续对表的操作都需要增加数据库名称作为前缀,如 db.tb。
// insert data
int affectedRows = stmt.executeUpdate("insert into tb values(now, 23, 10.3) (now + 1s, 20, 9.3)");
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 tb");
Timestamp ts = null;
int temperature = 0;
float humidity = 0;
while(resultSet.next()){
ts = resultSet.getTimestamp(1);
temperature = resultSet.getInt(2);
humidity = resultSet.getFloat("humidity");
System.out.printf("%s, %d, %s\n", ts, temperature, humidity);
}
查询和操作关系型数据库一致,使用下标获取返回字段内容时从 1 开始,建议使用字段名称获取。
在报错后,通过 SQLException 可以获取到错误的信息和错误码:
try (Statement statement = connection.createStatement()) {
// executeQuery
ResultSet resultSet = statement.executeQuery(sql);
// print result
printResult(resultSet);
} catch (SQLException e) {
System.out.println("ERROR Message: " + e.getMessage());
System.out.println("ERROR Code: " + e.getErrorCode());
e.printStackTrace();
}
JDBC 连接器可能报错的错误码包括 3 种:JDBC driver 本身的报错(错误码在 0x2301 到 0x2350 之间),原生连接方法的报错(错误码在 0x2351 到 0x2400 之间),TDengine 其他功能模块的报错。
具体的错误码请参考:
从 2.1.2.0 版本开始,TDengine 的 JDBC 原生连接实现大幅改进了参数绑定方式对数据写入(INSERT)场景的支持。采用这种方式写入数据时,能避免 SQL 语法解析的资源消耗,从而在很多情况下显著提升写入性能。
注意:
public class ParameterBindingDemo {
private static final String host = "127.0.0.1";
private static final Random random = new Random(System.currentTimeMillis());
private static final int BINARY_COLUMN_SIZE = 20;
private static final String[] schemaList = {
"create table stable1(ts timestamp, f1 tinyint, f2 smallint, f3 int, f4 bigint) tags(t1 tinyint, t2 smallint, t3 int, t4 bigint)",
"create table stable2(ts timestamp, f1 float, f2 double) tags(t1 float, t2 double)",
"create table stable3(ts timestamp, f1 bool) tags(t1 bool)",
"create table stable4(ts timestamp, f1 binary(" + BINARY_COLUMN_SIZE + ")) tags(t1 binary(" + BINARY_COLUMN_SIZE + "))",
"create table stable5(ts timestamp, f1 nchar(" + BINARY_COLUMN_SIZE + ")) tags(t1 nchar(" + BINARY_COLUMN_SIZE + "))"
};
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);
bindInteger(conn);
bindFloat(conn);
bindBoolean(conn);
bindBytes(conn);
bindString(conn);
conn.close();
}
private static void init(Connection conn) throws SQLException {
try (Statement stmt = conn.createStatement()) {
stmt.execute("drop database if exists test_parabind");
stmt.execute("create database if not exists test_parabind");
stmt.execute("use test_parabind");
for (int i = 0; i
}
}
}
private static void bindInteger(Connection conn) throws SQLException {
String sql = "insert into ? using stable1 tags(?,?,?,?) values(?,?,?,?,?)";
try (TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class)) {
for (int i = 1; i <= numOfSubTable; i++) {
// set table name
pstmt.setTableName("t1_" + i);
// set tags
pstmt.setTagByte(0, Byte.parseByte(Integer.toString(random.nextInt(Byte.MAX_VALUE))));
pstmt.setTagShort(1, Short.parseShort(Integer.toString(random.nextInt(Short.MAX_VALUE))));
pstmt.setTagInt(2, random.nextInt(Integer.MAX_VALUE));
pstmt.setTagLong(3, random.nextLong());
// set columns
ArrayList
long current = System.currentTimeMillis();
for (int j = 0; j
pstmt.setTimestamp(0, tsList);
ArrayList
for (int j = 0; j
pstmt.setByte(1, f1List);
ArrayList
for (int j = 0; j
pstmt.setShort(2, f2List);
ArrayList
for (int j = 0; j
pstmt.setInt(3, f3List);
ArrayList
for (int j = 0; j
pstmt.setLong(4, f4List);
// add column
pstmt.columnDataAddBatch();
}
// execute column
pstmt.columnDataExecuteBatch();
}
}
private static void bindFloat(Connection conn) throws SQLException {
String sql = "insert into ? using stable2 tags(?,?) values(?,?,?)";
TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class);
for (int i = 1; i <= numOfSubTable; i++) {
// set table name
pstmt.setTableName("t2_" + i);
// set tags
pstmt.setTagFloat(0, random.nextFloat());
pstmt.setTagDouble(1, random.nextDouble());
// set columns
ArrayList
long current = System.currentTimeMillis();
for (int j = 0; j
pstmt.setTimestamp(0, tsList);
ArrayList
for (int j = 0; j
pstmt.setFloat(1, f1List);
ArrayList
for (int j = 0; j
pstmt.setDouble(2, f2List);
// add column
pstmt.columnDataAddBatch();
}
// execute
pstmt.columnDataExecuteBatch();
// close if no try-with-catch statement is used
pstmt.close();
}
private static void bindBoolean(Connection conn) throws SQLException {
String sql = "insert into ? using stable3 tags(?) values(?,?)";
try (TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class)) {
for (int i = 1; i <= numOfSubTable; i++) {
// set table name
pstmt.setTableName("t3_" + i);
// set tags
pstmt.setTagBoolean(0, random.nextBoolean());
// set columns
ArrayList
long current = System.currentTimeMillis();
for (int j = 0; j
pstmt.setTimestamp(0, tsList);
ArrayList
for (int j = 0; j
pstmt.setBoolean(1, f1List);
// add column
pstmt.columnDataAddBatch();
}
// execute
pstmt.columnDataExecuteBatch();
}
}
private static void bindBytes(Connection conn) throws SQLException {
String sql = "insert into ? using stable4 tags(?) values(?,?)";
try (TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class)) {
for (int i = 1; i <= numOfSubTable; i++) {
// set table name
pstmt.setTableName("t4_" + i);
// set tags
pstmt.setTagString(0, new String("abc"));
// set columns
ArrayList
long current = System.currentTimeMillis();
for (int j = 0; j
pstmt.setTimestamp(0, tsList);
ArrayList
for (int j = 0; j
}
pstmt.setString(1, f1List, BINARY_COLUMN_SIZE);
// add column
pstmt.columnDataAddBatch();
}
// execute
pstmt.columnDataExecuteBatch();
}
}
private static void bindString(Connection conn) throws SQLException {
String sql = "insert into ? using stable5 tags(?) values(?,?)";
try (TSDBPreparedStatement pstmt = conn.prepareStatement(sql).unwrap(TSDBPreparedStatement.class)) {
for (int i = 1; i <= numOfSubTable; i++) {
// set table name
pstmt.setTableName("t5_" + i);
// set tags
pstmt.setTagNString(0, "北京-abc");
// set columns
ArrayList
long current = System.currentTimeMillis();
for (int j = 0; j
pstmt.setTimestamp(0, tsList);
ArrayList
for (int j = 0; j
}
pstmt.setNString(1, f1List, BINARY_COLUMN_SIZE);
// add column
pstmt.columnDataAddBatch();
}
// execute
pstmt.columnDataExecuteBatch();
}
}
}
用于设定 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)
用于设定 VALUES 数据列的取值的方法总共有:
public void setInt(int columnIndex, ArrayList
public void setFloat(int columnIndex, ArrayList
public void setTimestamp(int columnIndex, ArrayList
public void setLong(int columnIndex, ArrayList
public void setDouble(int columnIndex, ArrayList
public void setBoolean(int columnIndex, ArrayList
public void setByte(int columnIndex, ArrayList
public void setShort(int columnIndex, ArrayList
public void setString(int columnIndex, ArrayList
public void setNString(int columnIndex, ArrayList
从 2.2.0.0 版本开始,TDengine 增加了对无模式写入功能。无模式写入兼容 InfluxDB 的 行协议(Line Protocol)、OpenTSDB 的 telnet 行协议和 OpenTSDB 的 JSON 格式协议。详情请参见无模式写入。
注意:
public class SchemalessInsertTest {
private static final String host = "127.0.0.1";
private static final String lineDemo = "st,t1=3i64,t2=4f64,t3=\"t3\" c1=3i64,c3=L\"passit\",c2=false,c4=4f64 1626006833639000000";
private static final String telnetDemo = "stb0_0 1626006833 4 host=host0 interface=eth0";
private static final String jsOnDemo= "{\"metric\": \"meter_current\",\"timestamp\": 1346846400,\"value\": 10.3, \"tags\": {\"groupid\": 2, \"location\": \"Beijing\", \"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.executeUpdate("drop database if exists test_schemaless");
stmt.executeUpdate("create database if not exists test_schemaless");
stmt.executeUpdate("use test_schemaless");
}
}
}
TDengine Java 连接器支持订阅功能,应用 API 如下:
TSDBSubscribe sub = ((TSDBConnection)conn).subscribe("topic", "select * from meters", false);
subscribe
方法的三个参数含义如下:
select
语句,只应查询原始数据,只能按时间正序查询数据如上面的例子将使用 SQL 语句 select * from meters
创建一个名为 topic
的订阅,如果这个订阅已经存在,将继续之前的查询进度,而不是从头开始消费所有的数据。
int total = 0;
while(true) {
TSDBResultSet rs = sub.consume();
int count = 0;
while(rs.next()) {
count++;
}
total += count;
System.out.printf("%d rows consumed, total %d\n", count, total);
Thread.sleep(1000);
}
consume
方法返回一个结果集,其中包含从上次 consume
到目前为止的所有新数据。请务必按需选择合理的调用 consume
的频率(如例子中的 Thread.sleep(1000)
),否则会给服务端造成不必要的压力。
sub.close(true);
close
方法关闭一个订阅。如果其参数为 true
表示保留订阅进度信息,后续可以创建同名订阅继续消费数据;如为 false
则不保留订阅进度。
resultSet.close();
stmt.close();
conn.close();
注意务必要将 connection 进行关闭
,否则会出现连接泄露。
使用示例如下:
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
// ...
connection.close(); // put back to conneciton pool
}
通过 HikariDataSource.getConnection() 获取连接后,使用完成后需要调用 close() 方法,实际上它并不会关闭连接,只是放回连接池中。
更多 HikariCP 使用问题请查看官方说明。
使用示例如下:
public static void main(String[] args) throws Exception {
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
// ...
connection.close(); // put back to conneciton pool
}
更多 druid 使用问题请查看官方说明。
注意事项:
v1.6.4.1
版本开始提供了一个专门用于心跳检测的函数 select server_status()
,所以在使用连接池时推荐使用 select server_status()
进行 Validation Query。如下所示,select server_status()
执行成功会返回 1
。
taos> select server_status();
server_status()|
================
1 |
Query OK, 1 row(s) in set (0.000141s)
示例程序源码位于 TDengine/examples/JDBC
下:
请参考:JDBC example
taos-jdbcdriver 版本 | 主要变化 |
---|---|
2.0.38 | JDBC REST 连接增加批量拉取功能 |
2.0.37 | 增加对 json tag 支持 |
2.0.36 | 增加对 schemaless 写入支持 |
addBatch
和 executeBatch
来执行“批量写入/更行”,为什么没有带来性能上的提升?原因:TDengine 的 JDBC 实现中,通过 addBatch
方法提交的 SQL 语句,会按照添加的顺序,依次执行,这种方式没有减少与服务端的交互次数,不会带来性能上的提升。
解决方法:1. 在一条 insert 语句中拼接多个 values 值;2. 使用多线程的方式并发插入;3. 使用参数绑定的写入方式
原因:程序没有找到依赖的本地函数库 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
即可。
原因:目前 TDengine 只支持 64 位 JDK。
解决方法:重新安装 64 位 JDK。
taos-jdbcdriver doc