From 4135d0851908274a41f41df527cdf178e4bf0226 Mon Sep 17 00:00:00 2001 From: sheyanjie-qq <249478495@qq.com> Date: Fri, 2 Aug 2024 14:50:07 +0800 Subject: [PATCH] mod java code --- .../rust/nativeexample/examples/query.rs | 97 +++-- .../rust/nativeexample/examples/tmq.rs | 8 +- .../examples/rust/restexample/examples/tmq.rs | 10 +- docs/zh/08-develop/07-tmq.md | 5 +- .../com/taosdata/example/AbsConsumerLoop.java | 146 ++++---- .../taosdata/example/ConsumerLoopFull.java | 209 +++++------ .../com/taosdata/example/ConsumerLoopImp.java | 6 +- .../taosdata/example/ConsumerOffsetSeek.java | 42 +-- .../com/taosdata/example/JdbcBasicDemo.java | 118 +++--- .../com/taosdata/example/JdbcCreatDBDemo.java | 48 +-- .../taosdata/example/JdbcInsertDataDemo.java | 50 +-- .../com/taosdata/example/JdbcQueryDemo.java | 50 +-- .../com/taosdata/example/JdbcReqIdDemo.java | 38 +- .../taosdata/example/WsConsumerLoopFull.java | 337 +++++++++--------- 14 files changed, 592 insertions(+), 572 deletions(-) diff --git a/docs/examples/rust/nativeexample/examples/query.rs b/docs/examples/rust/nativeexample/examples/query.rs index a493e7c729..55c2feb7b7 100644 --- a/docs/examples/rust/nativeexample/examples/query.rs +++ b/docs/examples/rust/nativeexample/examples/query.rs @@ -7,62 +7,61 @@ async fn main() -> anyhow::Result<()> { let taos = builder.build()?; -// ANCHOR: create_db_and_table -let db = "power"; -// create database -taos.exec_many([ - format!("CREATE DATABASE IF NOT EXISTS `{db}`"), - format!("USE `{db}`"), -]) -.await?; -println!("Create database power successfully."); + // ANCHOR: create_db_and_table + let db = "power"; + // create database + taos.exec_many([ + format!("CREATE DATABASE IF NOT EXISTS `{db}`"), + format!("USE `{db}`"), + ]) + .await?; + println!("Create database power successfully."); -// create super table -taos.exec_many([ - "CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) \ - TAGS (`groupid` INT, `location` BINARY(24))", -]).await?; -println!("Create stable meters successfully."); + // create super table + taos.exec_many([ + "CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT, `phase` FLOAT) \ + TAGS (`groupid` INT, `location` BINARY(24))", + ]).await?; + println!("Create stable meters successfully."); -// ANCHOR_END: create_db_and_table + // ANCHOR_END: create_db_and_table -// ANCHOR: insert_data -let inserted = taos.exec("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) ").await?; + // ANCHOR: insert_data + let inserted = taos.exec("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) ").await?; -println!("inserted: {} rows to power.meters successfully.", inserted); -// ANCHOR_END: insert_data + println!("inserted: {} rows to power.meters successfully.", inserted); + // ANCHOR_END: insert_data -// ANCHOR: query_data -// query data, make sure the database and table are created before -let mut result = taos.query("SELECT ts, current, location FROM power.meters limit 100").await?; + // ANCHOR: query_data + // query data, make sure the database and table are created before + let mut result = taos.query("SELECT ts, current, location FROM power.meters limit 100").await?; -for field in result.fields() { - println!("got field: {}", field.name()); -} - -let mut rows = result.rows(); -let mut nrows = 0; -while let Some(row) = rows.try_next().await? { - for (col, (name, value)) in row.enumerate() { - println!( - "[{}] got value in col {} (named `{:>8}`): {}", - nrows, col, name, value - ); + for field in result.fields() { + println!("got field: {}", field.name()); } - nrows += 1; -} -// ANCHOR_END: query_data -// ANCHOR: query_with_req_id -let result = taos.query_with_req_id("SELECT ts, current, location FROM power.meters limit 1", 1).await?; -// ANCHOR_END: query_with_req_id + let mut rows = result.rows(); + let mut nrows = 0; + while let Some(row) = rows.try_next().await? { + for (col, (name, value)) in row.enumerate() { + println!( + "[{}] got value in col {} (named `{:>8}`): {}", + nrows, col, name, value + ); + } + nrows += 1; + } + // ANCHOR_END: query_data + // ANCHOR: query_with_req_id + let result = taos.query_with_req_id("SELECT ts, current, location FROM power.meters limit 1", 1).await?; + // ANCHOR_END: query_with_req_id } diff --git a/docs/examples/rust/nativeexample/examples/tmq.rs b/docs/examples/rust/nativeexample/examples/tmq.rs index 764c0c1fc8..04bf66b2ef 100644 --- a/docs/examples/rust/nativeexample/examples/tmq.rs +++ b/docs/examples/rust/nativeexample/examples/tmq.rs @@ -44,8 +44,12 @@ async fn main() -> anyhow::Result<()> { // ANCHOR_END: create_topic // ANCHOR: create_consumer - dsn.params.insert("group.id".to_string(), "abc".to_string()); - dsn.params.insert("auto.offset.reset".to_string(), "earliest".to_string()); + dsn.params.insert("auto.offset.reset".to_string(), "latest".to_string()); + dsn.params.insert("msg.with.table.name".to_string(), "true".to_string()); + dsn.params.insert("enable.auto.commit".to_string(), "true".to_string()); + dsn.params.insert("auto.commit.interval.ms".to_string(), "1000".to_string()); + dsn.params.insert("group.id".to_string(), "group1".to_string()); + dsn.params.insert("client.id".to_string(), "client1".to_string()); let builder = TmqBuilder::from_dsn(&dsn)?; let mut consumer = builder.build().await?; diff --git a/docs/examples/rust/restexample/examples/tmq.rs b/docs/examples/rust/restexample/examples/tmq.rs index 8f195e1629..0438e47515 100644 --- a/docs/examples/rust/restexample/examples/tmq.rs +++ b/docs/examples/rust/restexample/examples/tmq.rs @@ -9,9 +9,11 @@ async fn main() -> anyhow::Result<()> { .filter_level(log::LevelFilter::Info) .init(); use taos_query::prelude::*; + // ANCHOR: create_consumer_dsn let dsn = "ws://localhost:6041".to_string(); log::info!("dsn: {}", dsn); let mut dsn = Dsn::from_str(&dsn)?; + // ANCHOR_END: create_consumer_dsn let taos = TaosBuilder::from_dsn(&dsn)?.build().await?; @@ -41,8 +43,12 @@ async fn main() -> anyhow::Result<()> { // ANCHOR_END: create_topic // ANCHOR: create_consumer - dsn.params.insert("group.id".to_string(), "abc".to_string()); - dsn.params.insert("auto.offset.reset".to_string(), "earliest".to_string()); + dsn.params.insert("auto.offset.reset".to_string(), "latest".to_string()); + dsn.params.insert("msg.with.table.name".to_string(), "true".to_string()); + dsn.params.insert("enable.auto.commit".to_string(), "true".to_string()); + dsn.params.insert("auto.commit.interval.ms".to_string(), "1000".to_string()); + dsn.params.insert("group.id".to_string(), "group1".to_string()); + dsn.params.insert("client.id".to_string(), "client1".to_string()); let builder = TmqBuilder::from_dsn(&dsn)?; let mut consumer = builder.build().await?; diff --git a/docs/zh/08-develop/07-tmq.md b/docs/zh/08-develop/07-tmq.md index 405d227e8e..a5795b5b6e 100644 --- a/docs/zh/08-develop/07-tmq.md +++ b/docs/zh/08-develop/07-tmq.md @@ -104,9 +104,10 @@ Rust 连接器创建消费者的参数为 DSN， 可以设置的参数列表请 - ```rust -{{#include docs/examples/rust/nativeexample/examples/tmq.rs:create_consumer}} +{{#include docs/examples/rust/restexample/examples/tmq.rs:create_consumer_dsn}} + +{{#include docs/examples/rust/restexample/examples/tmq.rs:create_consumer}} ``` diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/AbsConsumerLoop.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/AbsConsumerLoop.java index f7570c742a..842abb4086 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/AbsConsumerLoop.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/AbsConsumerLoop.java @@ -24,24 +24,24 @@ public abstract class AbsConsumerLoop { public AbsConsumerLoop() throws SQLException { // ANCHOR: create_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", "client1"); -config.setProperty("value.deserializer", "com.taosdata.example.AbsConsumerLoop$ResultDeserializer"); -config.setProperty("value.deserializer.encoding", "UTF-8"); -try { - this.consumer = new TaosConsumer<>(config); -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to create jni consumer with " + config.getProperty("bootstrap.servers") + " ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to create consumer", ex); -} + 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", "client1"); + config.setProperty("value.deserializer", "com.taosdata.example.AbsConsumerLoop$ResultDeserializer"); + config.setProperty("value.deserializer.encoding", "UTF-8"); + try { + this.consumer = new TaosConsumer<>(config); + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to create jni consumer with " + config.getProperty("bootstrap.servers") + " ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to create consumer", ex); + } // ANCHOR_END: create_consumer this.topics = Collections.singletonList("topic_meters"); @@ -53,65 +53,65 @@ try { public void pollDataCodePiece() throws SQLException { // ANCHOR: poll_data_code_piece -try { - consumer.subscribe(topics); - while (!shutdown.get()) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - // process your data here - process(bean); + try { + consumer.subscribe(topics); + while (!shutdown.get()) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + // process your data here + process(bean); + } + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to poll data; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to poll data", ex); + } finally { + consumer.close(); + shutdownLatch.countDown(); } - } -} catch (SQLException ex){ - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to poll data; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to poll data", ex); -} finally { - consumer.close(); - shutdownLatch.countDown(); -} // ANCHOR_END: poll_data_code_piece } public void commitCodePiece() throws SQLException { // ANCHOR: commit_code_piece -try { - consumer.subscribe(topics); - while (!shutdown.get()) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - // process your data here - process(bean); + try { + consumer.subscribe(topics); + while (!shutdown.get()) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + // process your data here + process(bean); + } + if (!records.isEmpty()) { + // after processing the data, commit the offset manually + consumer.commitSync(); + } + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to execute consumer functions. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to execute consumer functions", ex); + } finally { + consumer.close(); + shutdownLatch.countDown(); } - if (!records.isEmpty()) { - // after processing the data, commit the offset manually - consumer.commitSync(); - } - } -} catch (SQLException ex){ - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to execute consumer functions. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to execute consumer functions", ex); -} finally { - consumer.close(); - shutdownLatch.countDown(); -} // ANCHOR_END: commit_code_piece } public void unsubscribeCodePiece() throws SQLException { // ANCHOR: unsubscribe_data_code_piece -try { - consumer.unsubscribe(); -} catch (SQLException ex){ - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to unsubscribe consumer. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to unsubscribe consumer", ex); -} finally { - consumer.close(); -} + try { + consumer.unsubscribe(); + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to unsubscribe consumer. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to unsubscribe consumer", ex); + } finally { + consumer.close(); + } // ANCHOR_END: unsubscribe_data_code_piece } @@ -119,14 +119,14 @@ try { try { // ANCHOR: poll_data -consumer.subscribe(topics); -while (!shutdown.get()) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - process(bean); - } -} + consumer.subscribe(topics); + while (!shutdown.get()) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + process(bean); + } + } // ANCHOR_END: poll_data consumer.unsubscribe(); diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopFull.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopFull.java index 4012f2e679..c78a875ab6 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopFull.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopFull.java @@ -15,37 +15,38 @@ import java.util.concurrent.TimeUnit; public class ConsumerLoopFull { static private Connection connection; static private Statement statement; - public static TaosConsumer getConsumer() throws SQLException{ -// ANCHOR: create_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("td.connect.user", "root"); -config.setProperty("td.connect.pass", "taosdata"); -config.setProperty("value.deserializer", "com.taosdata.example.ConsumerLoopFull$ResultDeserializer"); -config.setProperty("value.deserializer.encoding", "UTF-8"); -try { - return new TaosConsumer<>(config); -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to create jni consumer, host : " + config.getProperty("bootstrap.servers") + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to create consumer", ex); -} catch (Exception e) { - e.printStackTrace(); - throw new SQLException("Failed to create consumer", e); -} + public static TaosConsumer getConsumer() throws SQLException { +// ANCHOR: create_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("td.connect.user", "root"); + config.setProperty("td.connect.pass", "taosdata"); + config.setProperty("value.deserializer", "com.taosdata.example.ConsumerLoopFull$ResultDeserializer"); + config.setProperty("value.deserializer.encoding", "UTF-8"); + + try { + return new TaosConsumer<>(config); + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to create jni consumer, host : " + config.getProperty("bootstrap.servers") + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to create consumer", ex); + } catch (Exception e) { + e.printStackTrace(); + throw new SQLException("Failed to create consumer", e); + } // ANCHOR_END: create_consumer } public static void pollDataExample() throws SQLException { - try (TaosConsumer consumer = getConsumer()){ + try (TaosConsumer consumer = getConsumer()) { // subscribe to the topics List topics = Collections.singletonList("topic_meters"); @@ -72,96 +73,97 @@ try { public static void pollExample() throws SQLException { // ANCHOR: poll_data_code_piece -try (TaosConsumer consumer = getConsumer()){ - List topics = Collections.singletonList("topic_meters"); + try (TaosConsumer consumer = getConsumer()) { + List topics = Collections.singletonList("topic_meters"); - // subscribe to the topics - consumer.subscribe(topics); - System.out.println("subscribe topics successfully"); - for (int i = 0; i < 50; i++) { - // poll data - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - // process the data here - System.out.println("data: " + JSON.toJSONString(bean)); + // subscribe to the topics + consumer.subscribe(topics); + System.out.println("subscribe topics successfully"); + for (int i = 0; i < 50; i++) { + // poll data + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + // process the data here + System.out.println("data: " + JSON.toJSONString(bean)); + } + } + + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to poll data; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to poll data", ex); } - } - -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to poll data; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to poll data", ex); -} // ANCHOR_END: poll_data_code_piece -} + } public static void seekExample() throws SQLException { // ANCHOR: consumer_seek -try (TaosConsumer consumer = getConsumer()){ - List topics = Collections.singletonList("topic_meters"); + try (TaosConsumer consumer = getConsumer()) { + List topics = Collections.singletonList("topic_meters"); - // subscribe to the topics - consumer.subscribe(topics); - System.out.println("subscribe topics successfully"); - ConsumerRecords records = ConsumerRecords.emptyRecord(); - // make sure we have got some data - while (records.isEmpty()){ - records = consumer.poll(Duration.ofMillis(100)); - } + // subscribe to the topics + consumer.subscribe(topics); + System.out.println("subscribe topics successfully"); + ConsumerRecords records = ConsumerRecords.emptyRecord(); + // make sure we have got some data + while (records.isEmpty()) { + records = consumer.poll(Duration.ofMillis(100)); + } - for (ConsumerRecord record : records) { - System.out.println("first data polled: " + JSON.toJSONString(record.value())); - Set assignment = consumer.assignment(); - // seek to the beginning of the all partitions - consumer.seekToBeginning(assignment); - System.out.println("assignment seek to beginning successfully"); - break; - } + for (ConsumerRecord record : records) { + System.out.println("first data polled: " + JSON.toJSONString(record.value())); + Set assignment = consumer.assignment(); + // seek to the beginning of the all partitions + consumer.seekToBeginning(assignment); + System.out.println("assignment seek to beginning successfully"); + break; + } - // poll data agagin - records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - // process the data here - System.out.println("second data polled: " + JSON.toJSONString(record.value())); - break; - } + // poll data agagin + records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + // process the data here + System.out.println("second data polled: " + JSON.toJSONString(record.value())); + break; + } -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("seek example failed; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("seek example failed", ex); -} + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("seek example failed; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("seek example failed", ex); + } // ANCHOR_END: consumer_seek } public static void commitExample() throws SQLException { // ANCHOR: commit_code_piece -try (TaosConsumer consumer = getConsumer()){ - List topics = Collections.singletonList("topic_meters"); + try (TaosConsumer consumer = getConsumer()) { + List topics = Collections.singletonList("topic_meters"); - consumer.subscribe(topics); - for (int i = 0; i < 50; i++) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - // process your data here - System.out.println("data: " + JSON.toJSONString(bean)); + consumer.subscribe(topics); + for (int i = 0; i < 50; i++) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + // process your data here + System.out.println("data: " + JSON.toJSONString(bean)); + } + if (!records.isEmpty()) { + // after processing the data, commit the offset manually + consumer.commitSync(); + } + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to execute consumer functions. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to execute consumer functions", ex); } - if (!records.isEmpty()) { - // after processing the data, commit the offset manually - consumer.commitSync(); - } - } -} catch (SQLException ex){ - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to execute consumer functions. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to execute consumer functions", ex); -} // ANCHOR_END: commit_code_piece } + public static void unsubscribeExample() throws SQLException { TaosConsumer consumer = getConsumer(); List topics = Collections.singletonList("topic_meters"); @@ -169,7 +171,7 @@ try (TaosConsumer consumer = getConsumer()){ // ANCHOR: unsubscribe_data_code_piece try { consumer.unsubscribe(); - } catch (SQLException ex){ + } catch (SQLException ex) { // handle any errors, please refer to the JDBC specifications for detailed exceptions info System.out.println("Failed to unsubscribe consumer. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); throw new SQLException("Failed to unsubscribe consumer", ex); @@ -182,6 +184,7 @@ try (TaosConsumer consumer = getConsumer()){ public static class ResultDeserializer extends ReferenceDeserializer { } + // use this class to define the data structure of the result record public static class ResultBean { private Timestamp ts; @@ -240,12 +243,12 @@ try (TaosConsumer consumer = getConsumer()){ } } - public static void prepareData() throws SQLException{ + public static void prepareData() throws SQLException { StringBuilder insertQuery = new StringBuilder(); insertQuery.append("INSERT INTO " + "power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " + "VALUES "); - for (int i = 0; i < 10000; i++){ + for (int i = 0; i < 10000; i++) { insertQuery.append("(NOW + ").append(i).append("a, 10.30000, 219, 0.31000) "); } try { @@ -256,7 +259,8 @@ try (TaosConsumer consumer = getConsumer()){ throw new SQLException("Failed to insert data to power.meters", ex); } } - public static void prepareMeta() throws SQLException{ + + public static void prepareMeta() throws SQLException { try { statement.executeUpdate("CREATE DATABASE IF NOT EXISTS power"); statement.executeUpdate("USE power"); @@ -288,6 +292,7 @@ try (TaosConsumer consumer = getConsumer()){ } System.out.println("Connection created successfully."); } + public static void closeConnection() throws SQLException { try { if (statement != null) { @@ -337,16 +342,16 @@ try (TaosConsumer consumer = getConsumer()){ System.out.println("Data prepared successfully"); - // 关闭线程池，不再接收新任务 + // close the executor, which will make the executor reject new tasks executor.shutdown(); try { - // 等待直到所有任务完成 + // wait for the executor to terminate boolean result = executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); assert result; } catch (InterruptedException e) { Thread.currentThread().interrupt(); - } catch (Exception e){ + } catch (Exception e) { e.printStackTrace(); System.out.println("Wait executor termination failed."); } diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopImp.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopImp.java index 22f96841e6..84d84f062b 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopImp.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopImp.java @@ -20,9 +20,9 @@ public class ConsumerLoopImp { properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8"); // ANCHOR: create_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"); + 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"); // ANCHOR_END: create_topic statement.close(); diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerOffsetSeek.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerOffsetSeek.java index b9463d30f7..73901aba49 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerOffsetSeek.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerOffsetSeek.java @@ -35,29 +35,29 @@ public class ConsumerOffsetSeek { config.setProperty("value.deserializer.encoding", "UTF-8"); // ANCHOR: consumer_seek -String topic = "topic_meters"; -Map offset = null; -try (TaosConsumer 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 entry : offset.entrySet()) { - consumer.seek(entry.getKey(), entry.getValue()); + String topic = "topic_meters"; + Map offset = null; + try (TaosConsumer 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 entry : offset.entrySet()) { + consumer.seek(entry.getKey(), entry.getValue()); + } + } + ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); + // you can handle data here } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to execute consumer functions. server: " + config.getProperty("bootstrap.servers") + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to execute consumer functions", ex); } - ConsumerRecords records = consumer.poll(Duration.ofMillis(500)); - // you can handle data here - } -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to execute consumer functions. server: " + config.getProperty("bootstrap.servers") + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to execute consumer functions", ex); -} // ANCHOR_END: consumer_seek } } \ No newline at end of file diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcBasicDemo.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcBasicDemo.java index 4831567eab..945c321260 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcBasicDemo.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcBasicDemo.java @@ -16,97 +16,97 @@ public class JdbcBasicDemo { public static void main(String[] args) throws SQLException { -final String url = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; + final String url = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; // get connection -Properties properties = new Properties(); -properties.setProperty("charset", "UTF-8"); -properties.setProperty("locale", "en_US.UTF-8"); -properties.setProperty("timezone", "UTC-8"); -System.out.println("get connection starting..."); -try(Connection connection = DriverManager.getConnection(url, properties)){ + Properties properties = new Properties(); + properties.setProperty("charset", "UTF-8"); + properties.setProperty("locale", "en_US.UTF-8"); + properties.setProperty("timezone", "UTC-8"); + System.out.println("get connection starting..."); + try (Connection connection = DriverManager.getConnection(url, properties)) { -if (connection != null){ - System.out.println("[ OK ] Connection established."); -} else { - System.out.println("[ ERR ] Connection can not be established."); - return; -} + if (connection != null) { + System.out.println("[ OK ] Connection established."); + } else { + System.out.println("[ ERR ] Connection can not be established."); + return; + } -Statement stmt = connection.createStatement(); + Statement stmt = connection.createStatement(); // ANCHOR: create_db_and_table // create database -stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS power"); + stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS power"); // use database -stmt.executeUpdate("USE power"); + 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))"); + stmt.executeUpdate("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))"); // ANCHOR_END: create_db_and_table // ANCHOR: insert_data // 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."); + 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."); // ANCHOR_END: insert_data // ANCHOR: query_data // query data -ResultSet resultSet = stmt.executeQuery("SELECT * FROM meters"); + 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"); + 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); -} + System.out.printf("%s, %f, %s\n", ts, current, location); + } // ANCHOR_END: query_data // ANCHOR: with_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(); + 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(); // ANCHOR_END: with_reqid -String sql = "SELECT * FROM meters limit 2;"; + String sql = "SELECT * FROM meters limit 2;"; // ANCHOR: jdbc_exception -try (Statement statement = connection.createStatement(); - // executeQuery - ResultSet tempResultSet = statement.executeQuery(sql)) { + 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(); -} + // print result + printResult(tempResultSet); + } catch (SQLException e) { + System.out.println("ERROR Message: " + e.getMessage()); + System.out.println("ERROR Code: " + e.getErrorCode()); + e.printStackTrace(); + } // ANCHOR_END: jdbc_exception } catch (SQLException ex) { // handle any errors, please refer to the JDBC specifications for detailed exceptions info diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcCreatDBDemo.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcCreatDBDemo.java index d02b3c8789..f5f5b1c62c 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcCreatDBDemo.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcCreatDBDemo.java @@ -15,37 +15,37 @@ public class JdbcCreatDBDemo { public static void main(String[] args) throws SQLException { -final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; + final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; // get connection -Properties properties = new Properties(); -properties.setProperty("charset", "UTF-8"); -properties.setProperty("locale", "en_US.UTF-8"); -properties.setProperty("timezone", "UTC-8"); -System.out.println("get connection starting..."); + Properties properties = new Properties(); + properties.setProperty("charset", "UTF-8"); + properties.setProperty("locale", "en_US.UTF-8"); + properties.setProperty("timezone", "UTC-8"); + System.out.println("get connection starting..."); // ANCHOR: create_db_and_table -try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); - Statement stmt = connection.createStatement()) { + try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); + Statement stmt = connection.createStatement()) { - // create database - int rowsAffected = stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS power"); - // you can check rowsAffected here - assert rowsAffected == 0; + // create database + int rowsAffected = stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS power"); + // you can check rowsAffected here + assert rowsAffected == 0; - // use database - rowsAffected = stmt.executeUpdate("USE power"); - // you can check rowsAffected here - assert rowsAffected == 0; + // use database + rowsAffected = stmt.executeUpdate("USE power"); + // you can check rowsAffected here + assert rowsAffected == 0; - // create table - rowsAffected = stmt.executeUpdate("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))"); - // you can check rowsAffected here - assert rowsAffected == 0; + // create table + rowsAffected = stmt.executeUpdate("CREATE STABLE IF NOT EXISTS meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))"); + // you can check rowsAffected here + assert rowsAffected == 0; -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to create db and table, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); -} + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to create db and table, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + } // ANCHOR_END: create_db_and_table } diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcInsertDataDemo.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcInsertDataDemo.java index 7fc6d5137d..f8e11a8f41 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcInsertDataDemo.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcInsertDataDemo.java @@ -15,36 +15,36 @@ public class JdbcInsertDataDemo { public static void main(String[] args) throws SQLException { -final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; + final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; // get connection -Properties properties = new Properties(); -properties.setProperty("charset", "UTF-8"); -properties.setProperty("locale", "en_US.UTF-8"); -properties.setProperty("timezone", "UTC-8"); -System.out.println("get connection starting..."); + Properties properties = new Properties(); + properties.setProperty("charset", "UTF-8"); + properties.setProperty("locale", "en_US.UTF-8"); + properties.setProperty("timezone", "UTC-8"); + System.out.println("get connection starting..."); // ANCHOR: insert_data -try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); - Statement stmt = connection.createStatement()) { + try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); + Statement stmt = connection.createStatement()) { - // insert data, please make sure the database and table are created before - 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); - // you can check affectedRows here - System.out.println("inserted into " + affectedRows + " rows to power.meters successfully."); -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to insert data to power.meters, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + // insert data, please make sure the database and table are created before + 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); + // you can check affectedRows here + System.out.println("inserted into " + affectedRows + " rows to power.meters successfully."); + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to insert data to power.meters, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); -} + } // ANCHOR_END: insert_data } } diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcQueryDemo.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcQueryDemo.java index 30a835bb11..a4749afed5 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcQueryDemo.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcQueryDemo.java @@ -15,36 +15,36 @@ public class JdbcQueryDemo { public static void main(String[] args) throws SQLException { -final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; + final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; // get connection -Properties properties = new Properties(); -properties.setProperty("charset", "UTF-8"); -properties.setProperty("locale", "en_US.UTF-8"); -properties.setProperty("timezone", "UTC-8"); -System.out.println("get connection starting..."); + Properties properties = new Properties(); + properties.setProperty("charset", "UTF-8"); + properties.setProperty("locale", "en_US.UTF-8"); + properties.setProperty("timezone", "UTC-8"); + System.out.println("get connection starting..."); // ANCHOR: query_data -try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); - Statement stmt = connection.createStatement(); - // query data, make sure the database and table are created before - ResultSet resultSet = stmt.executeQuery("SELECT ts, current, location FROM power.meters limit 100")) { + try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); + Statement stmt = connection.createStatement(); + // query data, make sure the database and table are created before + ResultSet resultSet = stmt.executeQuery("SELECT ts, current, location FROM power.meters limit 100")) { - Timestamp ts; - float current; - String location; - while (resultSet.next()) { - ts = resultSet.getTimestamp(1); - current = resultSet.getFloat(2); - // we recommend using the column name to get the value - location = resultSet.getString("location"); + Timestamp ts; + float current; + String location; + while (resultSet.next()) { + ts = resultSet.getTimestamp(1); + current = resultSet.getFloat(2); + // we recommend using the column name to get the value + location = resultSet.getString("location"); - // you can check data here - System.out.printf("ts: %s, current: %f, location: %s %n", ts, current, location); - } -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to query data from power.meters, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); -} + // you can check data here + System.out.printf("ts: %s, current: %f, location: %s %n", ts, current, location); + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to query data from power.meters, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + } // ANCHOR_END: query_data } diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcReqIdDemo.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcReqIdDemo.java index cc557c18c5..3705e0aac2 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcReqIdDemo.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcReqIdDemo.java @@ -15,31 +15,31 @@ public class JdbcReqIdDemo { public static void main(String[] args) throws SQLException { -final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; + final String jdbcUrl = "jdbc:TAOS://" + host + ":6030/?user=" + user + "&password=" + password; // get connection -Properties properties = new Properties(); -properties.setProperty("charset", "UTF-8"); -properties.setProperty("locale", "en_US.UTF-8"); -properties.setProperty("timezone", "UTC-8"); -System.out.println("get connection starting..."); + Properties properties = new Properties(); + properties.setProperty("charset", "UTF-8"); + properties.setProperty("locale", "en_US.UTF-8"); + properties.setProperty("timezone", "UTC-8"); + System.out.println("get connection starting..."); // ANCHOR: with_reqid -try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); - // Create a statement that allows specifying a request ID - AbstractStatement aStmt = (AbstractStatement) connection.createStatement()) { + try (Connection connection = DriverManager.getConnection(jdbcUrl, properties); + // Create a statement that allows specifying a request ID + AbstractStatement aStmt = (AbstractStatement) connection.createStatement()) { + + try (ResultSet rs = aStmt.executeQuery("SELECT ts, current, location FROM power.meters limit 1", 3L)) { + while (rs.next()) { + Timestamp timestamp = rs.getTimestamp(1); + System.out.println("timestamp = " + timestamp); + } + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to execute sql with reqId, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - try (ResultSet rs = aStmt.executeQuery("SELECT ts, current, location FROM power.meters limit 1", 3L)) { - while (rs.next()) { - Timestamp timestamp = rs.getTimestamp(1); - System.out.println("timestamp = " + timestamp); } - } -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to execute sql with reqId, url:" + jdbcUrl + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - -} // ANCHOR_END: with_reqid } diff --git a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/WsConsumerLoopFull.java b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/WsConsumerLoopFull.java index d8084eb9f6..6caa42b1c1 100644 --- a/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/WsConsumerLoopFull.java +++ b/examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/WsConsumerLoopFull.java @@ -18,37 +18,38 @@ import java.util.concurrent.TimeUnit; public class WsConsumerLoopFull { static private Connection connection; static private Statement statement; - public static TaosConsumer getConsumer() throws SQLException{ -// ANCHOR: create_consumer -Properties config = new Properties(); -config.setProperty("td.connect.type", "ws"); -config.setProperty("bootstrap.servers", "localhost:6041"); -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", "client1"); -config.setProperty("td.connect.user", "root"); -config.setProperty("td.connect.pass", "taosdata"); -config.setProperty("value.deserializer", "com.taosdata.example.WsConsumerLoopFull$ResultDeserializer"); -config.setProperty("value.deserializer.encoding", "UTF-8"); -try { - return new TaosConsumer<>(config); - } catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to create websocket consumer, host : " + config.getProperty("bootstrap.servers") + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to create consumer", ex); - } catch (Exception e) { - e.printStackTrace(); - throw new SQLException("Failed to create consumer", e); - } + public static TaosConsumer getConsumer() throws SQLException { +// ANCHOR: create_consumer + Properties config = new Properties(); + config.setProperty("td.connect.type", "ws"); + config.setProperty("bootstrap.servers", "localhost:6041"); + 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", "client1"); + config.setProperty("td.connect.user", "root"); + config.setProperty("td.connect.pass", "taosdata"); + config.setProperty("value.deserializer", "com.taosdata.example.WsConsumerLoopFull$ResultDeserializer"); + config.setProperty("value.deserializer.encoding", "UTF-8"); + + try { + return new TaosConsumer<>(config); + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to create websocket consumer, host : " + config.getProperty("bootstrap.servers") + "; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to create consumer", ex); + } catch (Exception e) { + e.printStackTrace(); + throw new SQLException("Failed to create consumer", e); + } // ANCHOR_END: create_consumer -} + } public static void pollDataExample() throws SQLException { - try (TaosConsumer consumer = getConsumer()){ + try (TaosConsumer consumer = getConsumer()) { // subscribe to the topics List topics = Collections.singletonList("topic_meters"); @@ -75,178 +76,180 @@ try { public static void pollExample() throws SQLException { // ANCHOR: poll_data_code_piece -try (TaosConsumer consumer = getConsumer()){ - List topics = Collections.singletonList("topic_meters"); + try (TaosConsumer consumer = getConsumer()) { + List topics = Collections.singletonList("topic_meters"); - // subscribe to the topics - consumer.subscribe(topics); - System.out.println("subscribe topics successfully"); - for (int i = 0; i < 50; i++) { - // poll data - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - // process the data here - System.out.println("data: " + JSON.toJSONString(bean)); + // subscribe to the topics + consumer.subscribe(topics); + System.out.println("subscribe topics successfully"); + for (int i = 0; i < 50; i++) { + // poll data + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + // process the data here + System.out.println("data: " + JSON.toJSONString(bean)); + } + } + + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to poll data; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to poll data", ex); } - } - -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to poll data; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to poll data", ex); -} // ANCHOR_END: poll_data_code_piece } public static void seekExample() throws SQLException { // ANCHOR: consumer_seek -try (TaosConsumer consumer = getConsumer()){ - List topics = Collections.singletonList("topic_meters"); + try (TaosConsumer consumer = getConsumer()) { + List topics = Collections.singletonList("topic_meters"); - // subscribe to the topics - consumer.subscribe(topics); - System.out.println("subscribe topics successfully"); - ConsumerRecords records = ConsumerRecords.emptyRecord(); - // make sure we have got some data - while (records.isEmpty()){ - records = consumer.poll(Duration.ofMillis(100)); - } + // subscribe to the topics + consumer.subscribe(topics); + System.out.println("subscribe topics successfully"); + ConsumerRecords records = ConsumerRecords.emptyRecord(); + // make sure we have got some data + while (records.isEmpty()) { + records = consumer.poll(Duration.ofMillis(100)); + } - for (ConsumerRecord record : records) { - System.out.println("first data polled: " + JSON.toJSONString(record.value())); - Set assignment = consumer.assignment(); - // seek to the beginning of the all partitions - consumer.seekToBeginning(assignment); - System.out.println("assignment seek to beginning successfully"); - break; - } + for (ConsumerRecord record : records) { + System.out.println("first data polled: " + JSON.toJSONString(record.value())); + Set assignment = consumer.assignment(); + // seek to the beginning of the all partitions + consumer.seekToBeginning(assignment); + System.out.println("assignment seek to beginning successfully"); + break; + } - // poll data agagin - records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - // process the data here - System.out.println("second data polled: " + JSON.toJSONString(record.value())); - break; - } -} catch (SQLException ex) { - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("seek example failed; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("seek example failed", ex); -} + // poll data agagin + records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + // process the data here + System.out.println("second data polled: " + JSON.toJSONString(record.value())); + break; + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("seek example failed; ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("seek example failed", ex); + } // ANCHOR_END: consumer_seek } public static void commitExample() throws SQLException { // ANCHOR: commit_code_piece -try (TaosConsumer consumer = getConsumer()){ - List topics = Collections.singletonList("topic_meters"); + try (TaosConsumer consumer = getConsumer()) { + List topics = Collections.singletonList("topic_meters"); - consumer.subscribe(topics); - for (int i = 0; i < 50; i++) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); - for (ConsumerRecord record : records) { - ResultBean bean = record.value(); - // process your data here - System.out.println("data: " + JSON.toJSONString(bean)); + consumer.subscribe(topics); + for (int i = 0; i < 50; i++) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : records) { + ResultBean bean = record.value(); + // process your data here + System.out.println("data: " + JSON.toJSONString(bean)); + } + if (!records.isEmpty()) { + // after processing the data, commit the offset manually + consumer.commitSync(); + } + } + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to execute consumer functions. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to execute consumer functions", ex); } - if (!records.isEmpty()) { - // after processing the data, commit the offset manually - consumer.commitSync(); - } - } -} catch (SQLException ex){ - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to execute consumer functions. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to execute consumer functions", ex); -} // ANCHOR_END: commit_code_piece } + public static void unsubscribeExample() throws SQLException { TaosConsumer consumer = getConsumer(); List topics = Collections.singletonList("topic_meters"); consumer.subscribe(topics); // ANCHOR: unsubscribe_data_code_piece -try { - consumer.unsubscribe(); -} catch (SQLException ex){ - // handle any errors, please refer to the JDBC specifications for detailed exceptions info - System.out.println("Failed to unsubscribe consumer. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); - throw new SQLException("Failed to unsubscribe consumer", ex); -} finally { - consumer.close(); -} + try { + consumer.unsubscribe(); + } catch (SQLException ex) { + // handle any errors, please refer to the JDBC specifications for detailed exceptions info + System.out.println("Failed to unsubscribe consumer. ErrCode:" + ex.getErrorCode() + "; ErrMessage: " + ex.getMessage()); + throw new SQLException("Failed to unsubscribe consumer", ex); + } finally { + consumer.close(); + } // ANCHOR_END: unsubscribe_data_code_piece } -public static class ResultDeserializer extends ReferenceDeserializer { + public static class ResultDeserializer extends ReferenceDeserializer { -} -// use this class to define the data structure of the result record -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; + // use this class to define the data structure of the result record + 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; + } } - 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; - } -} - - public static void prepareData() throws SQLException{ + public static void prepareData() throws SQLException { StringBuilder insertQuery = new StringBuilder(); insertQuery.append("INSERT INTO " + "power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') " + "VALUES "); - for (int i = 0; i < 10000; i++){ + for (int i = 0; i < 10000; i++) { insertQuery.append("(NOW + ").append(i).append("a, 10.30000, 219, 0.31000) "); } try { @@ -257,7 +260,8 @@ public static class ResultBean { throw new SQLException("Failed to insert data to power.meters", ex); } } - public static void prepareMeta() throws SQLException{ + + public static void prepareMeta() throws SQLException { try { statement.executeUpdate("CREATE DATABASE IF NOT EXISTS power"); statement.executeUpdate("USE power"); @@ -289,6 +293,7 @@ public static class ResultBean { } System.out.println("Connection created successfully."); } + public static void closeConnection() throws SQLException { try { if (statement != null) { @@ -338,16 +343,16 @@ public static class ResultBean { System.out.println("Data prepared successfully"); - // 关闭线程池，不再接收新任务 + // close the executor, which will make the executor reject new tasks executor.shutdown(); try { - // 等待直到所有任务完成 + // wait for the executor to terminate boolean result = executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); assert result; } catch (InterruptedException e) { Thread.currentThread().interrupt(); - } catch (Exception e){ + } catch (Exception e) { e.printStackTrace(); System.out.println("Wait executor termination failed."); }