diff --git a/docs/zh/14-reference/03-connector/java.mdx b/docs/zh/14-reference/03-connector/java.mdx index 88a36746717005e11339164e65775ce47c1de11d..38c250b1107b1c045df4579bb684706811d4aaff 100644 --- a/docs/zh/14-reference/03-connector/java.mdx +++ b/docs/zh/14-reference/03-connector/java.mdx @@ -670,55 +670,127 @@ public class SchemalessInsertTest { TDengine Java 连接器支持订阅功能，应用 API 如下： -#### 创建订阅 +#### 创建 Topic ```java -TSDBSubscribe sub = ((TSDBConnection)conn).subscribe("topic", "select * from meters", false); +Connection connection = DriverManager.getConnection(url, properties); +Statement statement = connection.createStatement(); +statement.executeUpdate("create topic if not exists topic_speed as select ts, speed from speed_table"); ``` `subscribe` 方法的三个参数含义如下： -- topic：订阅的主题（即名称），此参数是订阅的唯一标识 -- sql：订阅的查询语句，此语句只能是 `select` 语句，只应查询原始数据，只能按时间正序查询数据 -- restart：如果订阅已经存在，是重新开始，还是继续之前的订阅 +- topic_speed：订阅的主题（即名称），此参数是订阅的唯一标识。 +- sql：订阅的查询语句，此语句只能是 `select` 语句，只应查询原始数据，只能按时间正序查询数据。 -如上面的例子将使用 SQL 语句 `select * from meters` 创建一个名为 `topic` 的订阅，如果这个订阅已经存在，将继续之前的查询进度，而不是从头开始消费所有的数据。 +如上面的例子将使用 SQL 语句 `select ts, speed from speed_table` 创建一个名为 `topic_speed` 的订阅。 + +#### 创建 Consumer + +```java +Properties config = new Properties(); +config.setProperty("enable.auto.commit", "true"); +config.setProperty("group.id", "group1"); +config.setProperty("value.deserializer", "com.taosdata.jdbc.tmq.ConsumerTest.ResultDeserializer"); + +TaosConsumer consumer = new TaosConsumer<>(config); +``` + +- enable.auto.commit: 是否允许自动提交。 +- group.id: consumer: 所在的 group。 +- value.deserializer: 结果集反序列化方法，可以继承 `com.taosdata.jdbc.tmq.ReferenceDeserializer`，并指定结果集 bean，实现反序列化。也可以继承 `com.taosdata.jdbc.tmq.Deserializer`，根据 SQL 的 resultSet 自定义反序列化方式。 +- 其他参数请参考：[Consumer 参数列表](/develop/tmq#创建-consumer-以及consumer-group) #### 订阅消费数据 ```java -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); + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ResultBean record : records) { + process(record); + } } ``` -`consume` 方法返回一个结果集，其中包含从上次 `consume` 到目前为止的所有新数据。请务必按需选择合理的调用 `consume` 的频率（如例子中的 `Thread.sleep(1000)`），否则会给服务端造成不必要的压力。 +`poll` 方法返回一个结果集，其中包含从上次 `poll` 到目前为止的所有新数据。请务必按需选择合理的调用 `poll` 的频率（如例子中的 `Duration.ofMillis(100)`），否则会给服务端造成不必要的压力。 #### 关闭订阅 ```java -sub.close(true); +consumer.close() ``` -`close` 方法关闭一个订阅。如果其参数为 `true` 表示保留订阅进度信息，后续可以创建同名订阅继续消费数据；如为 `false` 则不保留订阅进度。 - -### 关闭资源 +### 使用示例如下： ```java -resultSet.close(); -stmt.close(); -conn.close(); -``` +public abstract class ConsumerLoop { + private final TaosConsumer consumer; + private final List topics; + private final AtomicBoolean shutdown; + private final CountDownLatch shutdownLatch; + + public ConsumerLoop() throws SQLException { + Properties config = new Properties(); + config.setProperty("msg.with.table.name", "true"); + config.setProperty("enable.auto.commit", "true"); + config.setProperty("group.id", "group1"); + config.setProperty("value.deserializer", "com.taosdata.jdbc.tmq.ConsumerTest.ResultDeserializer"); + + this.consumer = new TaosConsumer<>(config); + this.topics = Collections.singletonList("topic_speed"); + 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 records = consumer.poll(Duration.ofMillis(100)); + for (ResultBean record : records) { + process(record); + } + } + } finally { + consumer.close(); + shutdownLatch.countDown(); + } + } + + public void shutdown() throws InterruptedException { + shutdown.set(true); + shutdownLatch.await(); + } + + static class ResultDeserializer extends ReferenceDeserializer { + + } + + static class ResultBean { + private Timestamp ts; + private int speed; + + public Timestamp getTs() { + return ts; + } -> `注意务必要将 connection 进行关闭`，否则会出现连接泄露。 + public void setTs(Timestamp ts) { + this.ts = ts; + } + + public int getSpeed() { + return speed; + } + + public void setSpeed(int speed) { + this.speed = speed; + } + } +} +``` ### 与连接池使用 diff --git a/source/client/src/TSDBJNIConnector.c b/source/client/src/TSDBJNIConnector.c index 227c2fff1860170f47074f7f3a7a73d1de569510..38bbec24ce32e633d20161ab815ac86423453192 100644 --- a/source/client/src/TSDBJNIConnector.c +++ b/source/client/src/TSDBJNIConnector.c @@ -757,7 +757,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_prepareStmtImp(J int32_t code = taos_stmt_prepare(pStmt, str, len); taosMemoryFreeClear(str); if (code != TSDB_CODE_SUCCESS) { - jniError("jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); + jniError("prepareStmt jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); return JNI_TDENGINE_ERROR; } @@ -785,7 +785,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_setBindTableNameI if (code != TSDB_CODE_SUCCESS) { (*env)->ReleaseStringUTFChars(env, jname, name); - jniError("jobj:%p, conn:%p, code:%s", jobj, tsconn, tstrerror(code)); + jniError("bindTableName jobj:%p, conn:%p, code:%s", jobj, tsconn, tstrerror(code)); return JNI_TDENGINE_ERROR; } @@ -860,7 +860,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_setTableNameTagsI (*env)->ReleaseStringUTFChars(env, tableName, name); if (code != TSDB_CODE_SUCCESS) { - jniError("jobj:%p, conn:%p, code:%s", jobj, tsconn, tstrerror(code)); + jniError("tableNameTags jobj:%p, conn:%p, code:%s", jobj, tsconn, tstrerror(code)); return JNI_TDENGINE_ERROR; } return JNI_SUCCESS; @@ -926,7 +926,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_bindColDataImp( taosMemoryFreeClear(b); if (code != TSDB_CODE_SUCCESS) { - jniError("jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); + jniError("bindColData jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); return JNI_TDENGINE_ERROR; } @@ -949,7 +949,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_addBatchImp(JNIEn int32_t code = taos_stmt_add_batch(pStmt); if (code != TSDB_CODE_SUCCESS) { - jniError("jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); + jniError("add batch jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); return JNI_TDENGINE_ERROR; } @@ -973,7 +973,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_executeBatchImp(J int32_t code = taos_stmt_execute(pStmt); if (code != TSDB_CODE_SUCCESS) { - jniError("jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); + jniError("excute batch jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); return JNI_TDENGINE_ERROR; } @@ -997,7 +997,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_closeStmt(JNIEnv int32_t code = taos_stmt_close(pStmt); if (code != TSDB_CODE_SUCCESS) { - jniError("jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); + jniError("close stmt jobj:%p, conn:%p, code:%s", jobj, tscon, tstrerror(code)); return JNI_TDENGINE_ERROR; }