Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
46be3845
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
未验证
提交
46be3845
编写于
8月 13, 2022
作者:
H
huolibo
提交者:
GitHub
8月 13, 2022
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
docs: add tmq sample (#16080)
上级
35768202
变更
6
显示空白变更内容
内联
并排
Showing
6 changed file
with
129 addition
and
45 deletion
+129
-45
docs/en/07-develop/_sub_java.mdx
docs/en/07-develop/_sub_java.mdx
+2
-0
docs/examples/java/pom.xml
docs/examples/java/pom.xml
+1
-1
docs/examples/java/src/main/java/com/taos/example/Meters.java
.../examples/java/src/main/java/com/taos/example/Meters.java
+62
-0
docs/examples/java/src/main/java/com/taos/example/MetersDeserializer.java
...va/src/main/java/com/taos/example/MetersDeserializer.java
+6
-0
docs/examples/java/src/main/java/com/taos/example/SubscribeDemo.java
...es/java/src/main/java/com/taos/example/SubscribeDemo.java
+56
-44
docs/zh/07-develop/_sub_java.mdx
docs/zh/07-develop/_sub_java.mdx
+2
-0
未找到文件。
docs/en/07-develop/_sub_java.mdx
浏览文件 @
46be3845
```java
{{#include docs/examples/java/src/main/java/com/taos/example/SubscribeDemo.java}}
{{#include docs/examples/java/src/main/java/com/taos/example/MetersDeserializer.java}}
{{#include docs/examples/java/src/main/java/com/taos/example/Meters.java}}
```
:::note
For now Java connector doesn't provide asynchronous subscription, but `TimerTask` can be used to achieve similar purpose.
...
...
docs/examples/java/pom.xml
浏览文件 @
46be3845
...
...
@@ -21,7 +21,7 @@
<dependency>
<groupId>
com.taosdata.jdbc
</groupId>
<artifactId>
taos-jdbcdriver
</artifactId>
<version>
2.0.38
</version>
<version>
3.0.0
</version>
</dependency>
<!-- ANCHOR_END: dep-->
<dependency>
...
...
docs/examples/java/src/main/java/com/taos/example/Meters.java
0 → 100644
浏览文件 @
46be3845
package
com.taos.example
;
import
java.sql.Timestamp
;
public
class
Meters
{
private
Timestamp
ts
;
private
float
current
;
private
int
voltage
;
private
int
groupid
;
private
String
location
;
public
Timestamp
getTs
()
{
return
ts
;
}
public
void
setTs
(
Timestamp
ts
)
{
this
.
ts
=
ts
;
}
public
float
getCurrent
()
{
return
current
;
}
public
void
setCurrent
(
float
current
)
{
this
.
current
=
current
;
}
public
int
getVoltage
()
{
return
voltage
;
}
public
void
setVoltage
(
int
voltage
)
{
this
.
voltage
=
voltage
;
}
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
;
}
@Override
public
String
toString
()
{
return
"Meters{"
+
"ts="
+
ts
+
", current="
+
current
+
", voltage="
+
voltage
+
", groupid="
+
groupid
+
", location='"
+
location
+
'\''
+
'}'
;
}
}
docs/examples/java/src/main/java/com/taos/example/MetersDeserializer.java
0 → 100644
浏览文件 @
46be3845
package
com.taos.example
;
import
com.taosdata.jdbc.tmq.ReferenceDeserializer
;
public
class
MetersDeserializer
extends
ReferenceDeserializer
<
Meters
>
{
}
\ No newline at end of file
docs/examples/java/src/main/java/com/taos/example/SubscribeDemo.java
浏览文件 @
46be3845
package
com.taos.example
;
import
com.taosdata.jdbc.TSDBConnection
;
import
com.taosdata.jdbc.TSDBDriver
;
import
com.taosdata.jdbc.TSDBResultSet
;
import
com.taosdata.jdbc.TSDBSubscribe
;
import
com.taosdata.jdbc.tmq.ConsumerRecords
;
import
com.taosdata.jdbc.tmq.TMQConstants
;
import
com.taosdata.jdbc.tmq.TaosConsumer
;
import
java.sql.Connection
;
import
java.sql.DriverManager
;
import
java.sql.ResultSetMetaData
;
import
java.sql.SQLException
;
import
java.sql.Statement
;
import
java.time.Duration
;
import
java.util.Collections
;
import
java.util.Properties
;
import
java.util.concurrent.TimeUnit
;
import
java.util.Timer
;
import
java.util.TimerTask
;
import
java.util.concurrent.atomic.AtomicBoolean
;
public
class
SubscribeDemo
{
private
static
final
String
topic
=
"topic-meter-current-bg-10"
;
private
static
final
String
sql
=
"select * from meters where current > 10"
;
private
static
final
String
TOPIC
=
"tmq_topic"
;
private
static
final
String
DB_NAME
=
"meters"
;
private
static
final
AtomicBoolean
shutdown
=
new
AtomicBoolean
(
false
);
public
static
void
main
(
String
[]
args
)
{
Connection
connection
=
null
;
TSDBSubscribe
subscribe
=
null
;
Timer
timer
=
new
Timer
();
timer
.
schedule
(
new
TimerTask
()
{
public
void
run
()
{
shutdown
.
set
(
true
);
}
},
3_000
);
try
{
// prepare
Class
.
forName
(
"com.taosdata.jdbc.TSDBDriver"
);
Properties
properties
=
new
Properties
();
properties
.
setProperty
(
TSDBDriver
.
PROPERTY_KEY_CHARSET
,
"UTF-8"
);
properties
.
setProperty
(
TSDBDriver
.
PROPERTY_KEY_TIME_ZONE
,
"UTC-8"
);
String
jdbcUrl
=
"jdbc:TAOS://127.0.0.1:6030/power?user=root&password=taosdata"
;
connection
=
DriverManager
.
getConnection
(
jdbcUrl
,
properties
);
// create subscribe
subscribe
=
((
TSDBConnection
)
connection
).
subscribe
(
topic
,
sql
,
true
);
int
count
=
0
;
while
(
count
<
10
)
{
// wait 1 second to avoid frequent calls to consume
TimeUnit
.
SECONDS
.
sleep
(
1
);
// consume
TSDBResultSet
resultSet
=
subscribe
.
consume
();
if
(
resultSet
==
null
)
{
continue
;
String
jdbcUrl
=
"jdbc:TAOS://127.0.0.1:6030/?user=root&password=taosdata"
;
Connection
connection
=
DriverManager
.
getConnection
(
jdbcUrl
);
try
(
Statement
statement
=
connection
.
createStatement
())
{
statement
.
executeUpdate
(
"drop topic if exists "
+
TOPIC
);
statement
.
executeUpdate
(
"drop database if exists "
+
DB_NAME
);
statement
.
executeUpdate
(
"create database "
+
DB_NAME
);
statement
.
executeUpdate
(
"use "
+
DB_NAME
);
statement
.
executeUpdate
(
"CREATE TABLE `meters` (`ts` TIMESTAMP, `current` FLOAT, `voltage` INT) TAGS (`groupid` INT, `location` BINARY(16))"
);
statement
.
executeUpdate
(
"CREATE TABLE `d0` USING `meters` TAGS(0, 'Los Angles')"
);
statement
.
executeUpdate
(
"INSERT INTO `d0` values(now - 10s, 0.32, 116)"
);
statement
.
executeUpdate
(
"INSERT INTO `d0` values(now - 8s, NULL, NULL)"
);
statement
.
executeUpdate
(
"INSERT INTO `d1` USING `meters` TAGS(1, 'San Francisco') values(now - 9s, 10.1, 119)"
);
statement
.
executeUpdate
(
"INSERT INTO `d1` values (now-8s, 10, 120) (now - 6s, 10, 119) (now - 4s, 11.2, 118)"
);
// create topic
statement
.
executeUpdate
(
"create topic "
+
TOPIC
+
" as select * from meters"
);
}
ResultSetMetaData
metaData
=
resultSet
.
getMetaData
();
while
(
resultSet
.
next
())
{
int
columnCount
=
metaData
.
getColumnCount
();
for
(
int
i
=
1
;
i
<=
columnCount
;
i
++)
{
System
.
out
.
print
(
metaData
.
getColumnLabel
(
i
)
+
": "
+
resultSet
.
getString
(
i
)
+
"\t"
);
// create consumer
Properties
properties
=
new
Properties
();
properties
.
setProperty
(
TMQConstants
.
BOOTSTRAP_SERVERS
,
"127.0.0.1:6030"
);
properties
.
setProperty
(
TMQConstants
.
MSG_WITH_TABLE_NAME
,
"true"
);
properties
.
setProperty
(
TMQConstants
.
ENABLE_AUTO_COMMIT
,
"true"
);
properties
.
setProperty
(
TMQConstants
.
GROUP_ID
,
"test"
);
properties
.
setProperty
(
TMQConstants
.
VALUE_DESERIALIZER
,
"com.taosdata.jdbc.MetersDeserializer"
);
// poll data
try
(
TaosConsumer
<
Meters
>
consumer
=
new
TaosConsumer
<>(
properties
))
{
consumer
.
subscribe
(
Collections
.
singletonList
(
TOPIC
));
while
(!
shutdown
.
get
())
{
ConsumerRecords
<
Meters
>
meters
=
consumer
.
poll
(
Duration
.
ofMillis
(
100
));
for
(
Meters
meter
:
meters
)
{
System
.
out
.
println
(
meter
);
}
System
.
out
.
println
();
count
++;
}
}
}
catch
(
Exception
e
)
{
}
catch
(
ClassNotFoundException
|
SQL
Exception
e
)
{
e
.
printStackTrace
();
}
finally
{
try
{
if
(
null
!=
subscribe
)
// close subscribe
subscribe
.
close
(
true
);
if
(
connection
!=
null
)
connection
.
close
();
}
catch
(
SQLException
throwable
)
{
throwable
.
printStackTrace
();
}
}
timer
.
cancel
();
}
}
\ No newline at end of file
docs/zh/07-develop/_sub_java.mdx
浏览文件 @
46be3845
```java
{{#include docs/examples/java/src/main/java/com/taos/example/SubscribeDemo.java}}
{{#include docs/examples/java/src/main/java/com/taos/example/MetersDeserializer.java}}
{{#include docs/examples/java/src/main/java/com/taos/example/Meters.java}}
```
:::note
目前 Java 接口没有提供异步订阅模式,但用户程序可以通过创建 `TimerTask` 等方式达到同样的效果。
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录