Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 223f8434 编写于 作者: D dingbo
浏览文件
操作

docs: java example

上级 0d3c845d
隐藏空白更改
内联 并排
Showing with 188 addition and 28 deletion
docs/examples/java/src/main/java/com/taos/example/highvolume/FastWriteExample.java
浏览文件 @ 223f8434
......@@ -58,7 +58,7 @@ class DataBaseMonitor {
public class FastWriteExample {
final static Logger logger = LoggerFactory.getLogger(FastWriteExample.class);
final static int taskQueueCapacity = Integer.MAX_VALUE / 100;
final static int taskQueueCapacity = 10000000;
final static List<BlockingQueue<String>> taskQueues = new ArrayList<>();
final static List<ReadTask> readTasks = new ArrayList<>();
final static List<WriteTask> writeTasks = new ArrayList<>();
......@@ -73,8 +73,12 @@ public class FastWriteExample {
public static void main(String[] args) throws InterruptedException, SQLException {
int readTaskCount = args.length > 0 ? Integer.parseInt(args[0]) : 1;
int writeTaskCount = args.length > 1 ? Integer.parseInt(args[1]) : 1;
logger.info("readTaskCount={}, writeTaskCount={}", readTaskCount, writeTaskCount);
int writeTaskCount = args.length > 1 ? Integer.parseInt(args[1]) : 3;
int tableCount = args.length > 2 ? Integer.parseInt(args[2]) : 1000;
int maxBatchSize = args.length > 3 ? Integer.parseInt(args[3]) : 3000;
logger.info("readTaskCount={}, writeTaskCount={} tableCount={} maxBatchSize={}",
readTaskCount, writeTaskCount, tableCount, maxBatchSize);
databaseMonitor.init().prepareDatabase();
......@@ -82,15 +86,16 @@ public class FastWriteExample {
for (int i = 0; i < writeTaskCount; ++i) {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(taskQueueCapacity);
taskQueues.add(queue);
WriteTask task = new WriteTask(queue);
WriteTask task = new WriteTask(queue, maxBatchSize);
Thread t = new Thread(task);
t.setName("WriteThread-" + i);
t.start();
}
// create reading tasks and start reading threads
int tableCountPerTask = tableCount / readTaskCount;
for (int i = 0; i < readTaskCount; ++i) {
ReadTask task = new ReadTask(i, taskQueues);
ReadTask task = new ReadTask(i, taskQueues, tableCountPerTask);
Thread t = new Thread(task);
t.setName("ReadThread-" + i);
t.start();
......@@ -102,7 +107,7 @@ public class FastWriteExample {
while (true) {
Thread.sleep(10000);
long count = databaseMonitor.count();
logger.info("total_count={} speed={}", count, (count - lastCount) / 10);
logger.info("count={} speed={}", count, (count - lastCount) / 10);
lastCount = count;
}
}
......
docs/examples/java/src/main/java/com/taos/example/highvolume/ReadTask.java
浏览文件 @ 223f8434
......@@ -12,7 +12,7 @@ import java.util.concurrent.BlockingQueue;
*/
class MockDataSource implements Iterator {
private String tbNamePrefix;
private int tableCount = 1000;
private int tableCount;
private long maxRowsPerTable = 1000000000L;
// 100 milliseconds between two neighbouring rows.
......@@ -26,8 +26,9 @@ class MockDataSource implements Iterator {
int[] voltage = {119, 116, 111, 113, 118};
float[] phase = {0.32f, 0.34f, 0.33f, 0.329f, 0.141f};
public MockDataSource(String tbNamePrefix) {
public MockDataSource(String tbNamePrefix, int tableCount) {
this.tbNamePrefix = tbNamePrefix;
this.tableCount = tableCount;
}
@Override
......@@ -62,12 +63,14 @@ class ReadTask implements Runnable {
private final int taskId;
private final List<BlockingQueue<String>> taskQueues;
private final int queueCount;
private final int tableCount;
private boolean active = true;
public ReadTask(int readTaskId, List<BlockingQueue<String>> queues) {
public ReadTask(int readTaskId, List<BlockingQueue<String>> queues, int tableCount) {
this.taskId = readTaskId;
this.taskQueues = queues;
this.queueCount = queues.size();
this.tableCount = tableCount;
}
/**
......@@ -87,7 +90,7 @@ class ReadTask implements Runnable {
@Override
public void run() {
logger.info("started");
Iterator<String> it = new MockDataSource("tb" + this.taskId);
Iterator<String> it = new MockDataSource("tb" + this.taskId, tableCount);
try {
while (it.hasNext() && active) {
String line = it.next();
......
docs/examples/java/src/main/java/com/taos/example/highvolume/SQLWriter.java
浏览文件 @ 223f8434
......@@ -17,7 +17,7 @@ import java.util.Map;
* <p>
* The main interfaces are two methods:
* <ol>
* <li>{@link SQLWriter#processLine}, which receive raw data from WriteTask and group raw data by table names.</li>
* <li>{@link SQLWriter#processLine}, which receive raw lines from WriteTask and group them by table names.</li>
* <li>{@link SQLWriter#flush}, which assemble INSERT statement and execute it.</li>
* </ol>
* <p>
......
docs/examples/java/src/main/java/com/taos/example/highvolume/WriteTask.java
浏览文件 @ 223f8434
......@@ -8,7 +8,7 @@ import java.util.concurrent.BlockingQueue;
// ANCHOR: WriteTask
class WriteTask implements Runnable {
private final static Logger logger = LoggerFactory.getLogger(WriteTask.class);
private final static int maxBatchSize = 3000;
private final int maxBatchSize;
// the queue from which this writing task get raw data.
private final BlockingQueue<String> queue;
......@@ -16,11 +16,11 @@ class WriteTask implements Runnable {
// A flag indicate whether to continue.
private boolean active = true;
public WriteTask(BlockingQueue<String> taskQueue) {
public WriteTask(BlockingQueue<String> taskQueue, int maxBatchSize) {
this.queue = taskQueue;
this.maxBatchSize = maxBatchSize;
}
@Override
public void run() {
logger.info("started");
......@@ -38,7 +38,7 @@ class WriteTask implements Runnable {
writer.flush();
} else {
// sleep a while to avoid high CPU usage if no more data in the queue and no buffered records, .
Thread.sleep(500);
Thread.sleep(100);
}
}
if (writer.hasBufferedValues()) {
......
docs/examples/java/src/main/resources/highvolume.drawio 0 → 100644
浏览文件 @ 223f8434
<mxfile host="65bd71144e">
<diagram id="_BjMg4p5x31hL4Gv-87s" name="第 1 页">
<mxGraphModel dx="862" dy="672" grid="0" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="900" pageHeight="900" background="none" math="0" shadow="0">
<root>
<mxCell id="0"/>
<mxCell id="1" parent="0"/>
<mxCell id="13" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;fontSize=17;" parent="1" source="7" edge="1">
<mxGeometry relative="1" as="geometry">
<mxPoint x="260" y="150" as="targetPoint"/>
</mxGeometry>
</mxCell>
<mxCell id="7" value="&lt;font style=&quot;font-size: 14px&quot;&gt;Read Task 1&lt;/font&gt;" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#dae8fc;strokeColor=#6c8ebf;fontStyle=1" parent="1" vertex="1">
<mxGeometry x="60" y="160" width="100" height="40" as="geometry"/>
</mxCell>
<mxCell id="16" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;entryX=0;entryY=0.75;entryDx=0;entryDy=0;fontSize=17;" parent="1" source="8" target="9" edge="1">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="18" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;entryX=0;entryY=0.5;entryDx=0;entryDy=0;fontSize=17;" parent="1" source="8" target="11" edge="1">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="8" value="&lt;font style=&quot;font-size: 14px&quot;&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;Read Tas&lt;/font&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;k 2&lt;/font&gt;&lt;/font&gt;" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#dae8fc;strokeColor=#6c8ebf;fontStyle=1" parent="1" vertex="1">
<mxGeometry x="60" y="240" width="100" height="40" as="geometry"/>
</mxCell>
<mxCell id="23" value="" style="edgeStyle=none;html=1;fontSize=14;" parent="1" source="9" target="20" edge="1">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="9" value="Queue 1" style="rounded=0;whiteSpace=wrap;html=1;fontSize=17;fillColor=#f8cecc;strokeColor=#b85450;" parent="1" vertex="1">
<mxGeometry x="265" y="130" width="245" height="30" as="geometry"/>
</mxCell>
<mxCell id="24" value="" style="edgeStyle=none;html=1;fontSize=14;" parent="1" source="10" target="21" edge="1">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="10" value="Queue 2" style="rounded=0;whiteSpace=wrap;html=1;fontSize=17;fillColor=#f8cecc;strokeColor=#b85450;" parent="1" vertex="1">
<mxGeometry x="265" y="200" width="245" height="30" as="geometry"/>
</mxCell>
<mxCell id="25" value="" style="edgeStyle=none;html=1;fontSize=14;" parent="1" source="11" target="22" edge="1">
<mxGeometry relative="1" as="geometry"/>
</mxCell>
<mxCell id="11" value="Queue 3" style="rounded=0;whiteSpace=wrap;html=1;fontSize=17;fillColor=#f8cecc;strokeColor=#b85450;" parent="1" vertex="1">
<mxGeometry x="265" y="280" width="245" height="30" as="geometry"/>
</mxCell>
<mxCell id="14" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;fontSize=17;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" parent="1" source="7" target="10" edge="1">
<mxGeometry relative="1" as="geometry">
<mxPoint x="270" y="160" as="targetPoint"/>
<mxPoint x="180" y="195" as="sourcePoint"/>
</mxGeometry>
</mxCell>
<mxCell id="15" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;fontSize=17;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" parent="1" source="7" target="11" edge="1">
<mxGeometry relative="1" as="geometry">
<mxPoint x="280" y="170" as="targetPoint"/>
<mxPoint x="190" y="205" as="sourcePoint"/>
</mxGeometry>
</mxCell>
<mxCell id="17" style="edgeStyle=none;html=1;exitX=1;exitY=0.5;exitDx=0;exitDy=0;entryX=-0.033;entryY=0.65;entryDx=0;entryDy=0;fontSize=17;entryPerimeter=0;" parent="1" source="8" target="10" edge="1">
<mxGeometry relative="1" as="geometry">
<mxPoint x="180" y="285" as="sourcePoint"/>
<mxPoint x="275" y="162.5" as="targetPoint"/>
</mxGeometry>
</mxCell>
<mxCell id="20" value="&lt;font style=&quot;font-size: 14px&quot;&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;Write Tas&lt;/font&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;k 2&lt;/font&gt;&lt;/font&gt;" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#dae8fc;strokeColor=#6c8ebf;fontStyle=1" parent="1" vertex="1">
<mxGeometry x="600" y="125" width="100" height="40" as="geometry"/>
</mxCell>
<mxCell id="21" value="&lt;font style=&quot;font-size: 14px&quot;&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;Write Tas&lt;/font&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;k 2&lt;/font&gt;&lt;/font&gt;" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#dae8fc;strokeColor=#6c8ebf;fontStyle=1" parent="1" vertex="1">
<mxGeometry x="600" y="195" width="100" height="40" as="geometry"/>
</mxCell>
<mxCell id="22" value="&lt;font style=&quot;font-size: 14px&quot;&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;Write Tas&lt;/font&gt;&lt;font style=&quot;font-size: 14px&quot;&gt;k 3&lt;/font&gt;&lt;/font&gt;" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#dae8fc;strokeColor=#6c8ebf;fontStyle=1" parent="1" vertex="1">
<mxGeometry x="600" y="275" width="100" height="40" as="geometry"/>
</mxCell>
</root>
</mxGraphModel>
</diagram>
</mxfile>
\ No newline at end of file
docs/zh/07-develop/03-insert-data/05-high-volume.md
浏览文件 @ 223f8434
......@@ -4,18 +4,20 @@ title: 高效写入
## 高效写入原理
本节介绍如何高效地向 TDengine 写入数据。高效写入数据要考虑几个因素：数据在不同表（或子表）之间的分布，即要写入数据的相邻性；单次写入的数据量；并发连接数。一般来说，每批次只向同一张表（或子表）写入数据比向多张表（或子表）写入数据要更高效；每批次写入的数据量越大越高效（但超过一定阈值其优势会消失）；同时写入数据的并发连接数越多写入越高效（但超过一定阈值反而会下降，取决于服务端处理能力）。
本节介绍如何高效地向 TDengine 写入数据。高效写入数据要考虑几个因素：数据在不同表（或子表）之间的分布，即要写入数据的相邻性；单次写入的数据量；并发连接数。一般来说，每批次只向同一张表（或子表）写入数据比向多张表（或子表）写入数据要更高效；每批次写入的数据量越大越高效（但超过一定阈值其优势会消失；同时写入数据的并发连接数越多写入越高效（但超过一定阈值反而会下降，取决于服务端处理能力）。
为了更高效地向 TDengine 写入数据，客户端程序要充分且恰当地利用以上几个因素。在单次写入中尽量只向同一张表（或子表）写入数据，每批次写入的数据量经过测试和调优设定为一个最适合当前系统处理能力的数值，并发写入的连接数同样经过测试和调优后设定为一个最适合当前系统处理能力的数值，以实现在当前系统中的最佳写入速度。同时，TDengine 还提供了独特的参数绑定写入，这也是一个有助于实现高效写入的方法。
## 高效写入方案
下面的示例程序展示了如何高效写入数据：
- TDengine 客户端程序从消息队列或者其它数据源不断读入数据，在示例程序中采用生成模拟数据的方式来模拟读取数据源
- 单个连接向 TDengine 写入的速度无法与读数据的速度相匹配，因此客户端程序启动多个线程，每个线程都建立了与 TDengine 的连接，每个线程都有一个独占的固定大小的消息队列
- 客户端程序将接收到的数据根据所属的表名（或子表名）HASH 到不同的线程，即写入该线程所对应的消息队列，以此确保属于某个表（或子表）的数据一定会被一个固定的线程处理
- 各个子线程在将所关联的消息队列中的数据读空后或者读取数据量达到一个预定的阈值后将该批数据写入 TDengine，并继续处理后面接收到的数据
![TDengine 高效写入线程模型](highvolume.webp)
## Java 示例程序
......@@ -23,18 +25,36 @@ title: 高效写入
### 主程序
主程序负责创建队列，并启动读线程和写线程。
主程序负责：
1. 创建消息队列
2. 启动写线程
3. 启动读线程
4. 每隔 10 秒统计一次写入速度
主程序默认暴露了 4 个参数，每次启动程序都可调节，用于测试和调优：
1. 读线程个数。默认为 1。
2. 写线程个数。默认为 3。
3. 模拟生成的总表数。默认为 1000。将会平分给各个读线程。
4. 每批最大数据量。默认为 3000。
```java title="主程序"
{{#include docs/examples/java/src/main/java/com/taos/example/highvolume/FastWriteExample.java:main}}
```
队列容量(taskQueueCapacity)也是与性能有关的参数，可通过修改程序调节。一般来讲，队列容量越大，入队被阻塞的概率越小，队列的吞吐量越大，但是内存占用也会越大。
### 读任务的实现
读任务负责从数据源读数据。每个读任务都关联了一个模拟数据源。每个模拟数据源可生成一点数量表的数据。不同的模拟数据源生成不同表的数据。
读任务采用阻塞的方式写消息队列。也就是说，一旦队列满了，写操作就会阻塞。
<details>
<summary>读任务的实现</summary>
```java
```java
{{#include docs/examples/java/src/main/java/com/taos/example/highvolume/ReadTask.java:ReadTask}}
```
......@@ -45,21 +65,45 @@ title: 高效写入
<details>
<summary>写任务的实现</summary>
```java
```java
{{#include docs/examples/java/src/main/java/com/taos/example/highvolume/WriteTask.java:WriteTask}}
```
</details>
### SQLWriter 类的实现
SQLWriter 类封装了拼 SQL 和写数据的逻辑。注意，所有的表都没有提前创建，而是写入出错的时候，再以超级表为模板建表，然后重现执行 INSERT 语句。
SQLWriter 类封装了拼 SQL 和写数据的逻辑。注意，所有的表都没有提前创建，而是写入出错的时候，再以超级表为模板批量建表，然后重新执行 INSERT 语句。
<details>
<summary>SQLWriter 类的实现</summary>
```java
{{#include docs/examples/java/src/main/java/com/taos/example/highvolume/SQLWriter.java:SQLWriter}}
```
</details>
### 执行示例程序
可用在本地集成开发环境里直接运行示例程序，注意要提前配置环境变量 TDENGINE_JDBC_URL。若要在远程服务器上执行示例程序，可按照下面步骤操作：
执行程序前需配置环境变量 `TDENGINE_JDBC_URL`。如果 TDengine Server 部署在本机，且用户名、密码和端口都是默认值，那么可配置：
```
TDENGINE_JDBC_URL="jdbc:TAOS://localhost:6030?user=root&password=taosdata"
```
若要在本地集成开发环境执行示例程序，只需：
1. clone TDengine 仓库
```
git clone git@github.com:taosdata/TDengine.git --depth 1
```
2. 用集成开发环境打开 `docs/examples/java` 目录。
3. 在开发环境中配置环境变量 `TDENGINE_JDBC_URL`。如果已配置了全局的环境变量 `TDENGINE_JDBC_URL` 可跳过这一步。
4. 运行类 `com.taos.example.highvolume.FastWriteExample`。
若要在服务器上执行示例程序，可按照下面的步骤操作：
1. 打包示例代码。在目录 TDengine/docs/examples/java 下执行：
```
mvn package
......@@ -68,23 +112,59 @@ SQLWriter 类封装了拼 SQL 和写数据的逻辑。注意，所有的表都
```
mkdir -p examples/java
```
3. 复制依赖到服务器指定目录，比如：
3. 复制依赖到服务器指定目录：
- 复制依赖包，只用复制一次
```
scp -r .\target\lib <user>@<host>:~/examples/java
```
- 复制本程序的 jar 包，每次更新代码都需要复制
```
scp -r .\target\javaexample-1.0.jar <user>@<host>:~/examples/java
```
4. 配置环境变量。
编辑 `~/.bash_profile` 或 `~/.bashrc` 添加如下内容例如：
```
scp -r .\target\lib user@host:~/examples/java
scp -r .\target\javaexample-1.0.jar user@host:~/examples/java
export TDENGINE_JDBC_URL="jdbc:TAOS://localhost:6030?user=root&password=taosdata"
```
4. 配置环境变量，例如：
以上使用的是本地部署 TDengine Server 时默认的 JDBC URL。你需要根据自己的实际情况更改。
5. 用 java 命令启动示例程序，命令模板：
```
export TDENGINE_JDBC_URL="jdbc:TAOS://localhost:6030?user=root&password=taosdata"
java -classpath lib/*:javaexample-1.0.jar com.taos.example.highvolume.FastWriteExample <read_thread_count> <white_thread_count> <total_table_count> <max_batch_size>
```
5. 用 java 命令启动示例程序
6. 结束测试程序。测试程序不会自动结束，在获取到当前配置下稳定的写入速度后，按 <kbd>CTRL</kbd> + <kbd>C</kbd> 结束程序。
下面是一次实际运行的截图：
```
java -classpath lib/*:javaexample-1.0.jar com.taos.example.highvolume.FastWriteExample <read_thread_count> <white_thread_count>
[bding@vm95 java]$ java -classpath lib/*:javaexample-1.0.jar com.taos.example.highvolume.FastWriteExample 1 9 1000 2000
17:01:01.131 [main] INFO c.t.e.highvolume.FastWriteExample - readTaskCount=1, writeTaskCount=9 tableCount=1000 maxBatchSize=2000
17:01:01.286 [WriteThread-0] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.354 [WriteThread-1] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.360 [WriteThread-2] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.366 [WriteThread-3] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.433 [WriteThread-4] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.438 [WriteThread-5] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.443 [WriteThread-6] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.448 [WriteThread-7] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.454 [WriteThread-8] INFO c.taos.example.highvolume.WriteTask - started
17:01:01.454 [ReadThread-0] INFO com.taos.example.highvolume.ReadTask - started
17:01:11.615 [main] INFO c.t.e.highvolume.FastWriteExample - count=18766442 speed=1876644
17:01:21.775 [main] INFO c.t.e.highvolume.FastWriteExample - count=38947464 speed=2018102
17:01:32.428 [main] INFO c.t.e.highvolume.FastWriteExample - count=58649571 speed=1970210
17:01:42.577 [main] INFO c.t.e.highvolume.FastWriteExample - count=79264890 speed=2061531
17:01:53.265 [main] INFO c.t.e.highvolume.FastWriteExample - count=99097476 speed=1983258
17:02:04.209 [main] INFO c.t.e.highvolume.FastWriteExample - count=119546779 speed=2044930
17:02:14.935 [main] INFO c.t.e.highvolume.FastWriteExample - count=141078914 speed=2153213
17:02:25.617 [main] INFO c.t.e.highvolume.FastWriteExample - count=162183457 speed=2110454
17:02:36.718 [main] INFO c.t.e.highvolume.FastWriteExample - count=182735614 speed=2055215
17:02:46.988 [main] INFO c.t.e.highvolume.FastWriteExample - count=202895614 speed=2016000
```
## Python 示例程序
在 Python 示例程序中采用参数绑定的写入方式。(开发中)
<!-- ```python title="Python 示例程序"
``` -->
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或