xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mod java code

This commit is contained in:
sheyanjie-qq 2024-08-02 14:50:07 +08:00 committed by gccgdb1234
parent ee5a8a96b0
commit 4135d08519
14 changed files with 592 additions and 572 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
docs/examples/rust/nativeexample/examples/query.rs
View File

@ -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
}

8
docs/examples/rust/nativeexample/examples/tmq.rs
View File

@ -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?;

10
docs/examples/rust/restexample/examples/tmq.rs
View File

@ -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?;

5
docs/zh/08-develop/07-tmq.md
View File

@ -104,9 +104,10 @@ Rust 连接器创建消费者的参数为 DSN， 可以设置的参数列表请
<TabItem label="Rust" value="rust">
```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}}
```
</TabItem>

146
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/AbsConsumerLoop.java
View File

@ -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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value();
// process your data here
process(bean);
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 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value();
// process your data here
process(bean);
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 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> record : records) {
ResultBean bean = record.value();
process(bean);
}
}
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);
}
}
// ANCHOR_END: poll_data
consumer.unsubscribe();

209
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopFull.java
View File

@ -15,37 +15,38 @@ import java.util.concurrent.TimeUnit;
public class ConsumerLoopFull {
static private Connection connection;
static private Statement statement;
public static TaosConsumer<ResultBean> 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<ResultBean> 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<ResultBean> consumer = getConsumer()){
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
// subscribe to the topics
List<String> topics = Collections.singletonList("topic_meters");
@ -72,96 +73,97 @@ try {
public static void pollExample() throws SQLException {
// ANCHOR: poll_data_code_piece
try (TaosConsumer<ResultBean> consumer = getConsumer()){
List<String> topics = Collections.singletonList("topic_meters");
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
List<String> 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> consumer = getConsumer()){
List<String> topics = Collections.singletonList("topic_meters");
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
List<String> topics = Collections.singletonList("topic_meters");
// subscribe to the topics
consumer.subscribe(topics);
System.out.println("subscribe topics successfully");
ConsumerRecords<ResultBean> 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<ResultBean> records = ConsumerRecords.emptyRecord();
// make sure we have got some data
while (records.isEmpty()) {
records = consumer.poll(Duration.ofMillis(100));
}
for (ConsumerRecord<ResultBean> record : records) {
System.out.println("first data polled: " + JSON.toJSONString(record.value()));
Set<TopicPartition> 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<ResultBean> record : records) {
System.out.println("first data polled: " + JSON.toJSONString(record.value()));
Set<TopicPartition> 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<ResultBean> 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<ResultBean> 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<ResultBean> consumer = getConsumer()){
List<String> topics = Collections.singletonList("topic_meters");
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
List<String> topics = Collections.singletonList("topic_meters");
consumer.subscribe(topics);
for (int i = 0; i < 50; i++) {
ConsumerRecords<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> consumer = getConsumer();
List<String> topics = Collections.singletonList("topic_meters");
@ -169,7 +171,7 @@ try (TaosConsumer<ResultBean> 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<ResultBean> consumer = getConsumer()){
public static class ResultDeserializer extends ReferenceDeserializer<ResultBean> {
}
// 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<ResultBean> 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<ResultBean> 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<ResultBean> consumer = getConsumer()){
}
System.out.println("Connection created successfully.");
}
public static void closeConnection() throws SQLException {
try {
if (statement != null) {
@ -337,16 +342,16 @@ try (TaosConsumer<ResultBean> 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.");
}

6
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerLoopImp.java
View File

@ -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();

42
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/ConsumerOffsetSeek.java
View File

@ -35,29 +35,29 @@ public class ConsumerOffsetSeek {
config.setProperty("value.deserializer.encoding", "UTF-8");
// ANCHOR: consumer_seek
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());
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));
// 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<AbsConsumerLoop.ResultBean> 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
}
}

118
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcBasicDemo.java
View File

@ -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

48
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcCreatDBDemo.java
View File

@ -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
}

50
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcInsertDataDemo.java
View File

@ -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
}
}

50
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcQueryDemo.java
View File

@ -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
}

38
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/JdbcReqIdDemo.java
View File

@ -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
}

337
examples/JDBC/JDBCDemo/src/main/java/com/taosdata/example/WsConsumerLoopFull.java
View File

@ -18,37 +18,38 @@ import java.util.concurrent.TimeUnit;
public class WsConsumerLoopFull {
static private Connection connection;
static private Statement statement;
public static TaosConsumer<ResultBean> 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<ResultBean> 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<ResultBean> consumer = getConsumer()){
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
// subscribe to the topics
List<String> topics = Collections.singletonList("topic_meters");
@ -75,178 +76,180 @@ try {
public static void pollExample() throws SQLException {
// ANCHOR: poll_data_code_piece
try (TaosConsumer<ResultBean> consumer = getConsumer()){
List<String> topics = Collections.singletonList("topic_meters");
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
List<String> 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> consumer = getConsumer()){
List<String> topics = Collections.singletonList("topic_meters");
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
List<String> topics = Collections.singletonList("topic_meters");
// subscribe to the topics
consumer.subscribe(topics);
System.out.println("subscribe topics successfully");
ConsumerRecords<ResultBean> 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<ResultBean> records = ConsumerRecords.emptyRecord();
// make sure we have got some data
while (records.isEmpty()) {
records = consumer.poll(Duration.ofMillis(100));
}
for (ConsumerRecord<ResultBean> record : records) {
System.out.println("first data polled: " + JSON.toJSONString(record.value()));
Set<TopicPartition> 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<ResultBean> record : records) {
System.out.println("first data polled: " + JSON.toJSONString(record.value()));
Set<TopicPartition> 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<ResultBean> 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<ResultBean> 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<ResultBean> consumer = getConsumer()){
List<String> topics = Collections.singletonList("topic_meters");
try (TaosConsumer<ResultBean> consumer = getConsumer()) {
List<String> topics = Collections.singletonList("topic_meters");
consumer.subscribe(topics);
for (int i = 0; i < 50; i++) {
ConsumerRecords<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<ResultBean> 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<ResultBean> consumer = getConsumer();
List<String> 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<ResultBean> {
public static class ResultDeserializer extends ReferenceDeserializer<ResultBean> {
}
// 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.");
}