TDengine 高效写入Efficient Writing实战指南基于 JDBC 连接器构建百万级吞吐的数据写入管道【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址: https://gitcode.com/taosdata/tdengine导读本文围绕 TDengine 官方文档 Ingesting Data Efficiently 的核心内容深入讲解 TDengine 连接器提供的高效写入Efficient Writing功能它如何通过自动创建写入线程与专属队列、按子表分片缓存、批量发送等机制让应用无需掌握多线程编程与数据分片技巧即可获得百万级写入吞吐。读者学完本文将掌握 JDBC 连接器高效写入的启用方式、核心配置参数、写入成功校验手段能基于仓库中的highvolume示例程序搭建模拟数据与 Kafka 两种高吞吐写入管道并从客户端、数据源、服务器三个视角系统化调优写入性能。一、什么是高效写入降低高吞吐写入的门槛构建百万级吞吐的数据写入管道传统做法要求开发者自己掌握多线程编程与数据分片sharding技术为不同的子表分配不同的写入线程、为每个线程维护缓冲区、在达到阈值或超时后批量发送……这既是技术门槛也是容易出错的地方。TDengine 连接器为此提供了**高效写入Efficient Writing**功能。启用后连接器会自动完成以下工作自动创建写入线程与专属队列根据配置参数动态创建独立的写入线程和对应的写入队列按子表分片缓存应用写入的数据由连接器自动按子表sub-table分片缓存在对应队列中批量触发发送当队列数据量达到预设阈值、或达到预设等待超时时间避免延迟过大时触发批量发送。这种机制大幅减少了网络请求次数从而提升吞吐量用户无需掌握多线程编程或数据分片知识单线程调用写入接口即可获得接近传统多线程写入的性能。从版本历史看JDBC 驱动自v3.6.0起在 WebSocket 连接上提供该特性当前仓库中的 highvolume 示例工程pom.xml依赖的taos-jdbcdriver版本为3.8.3请确保你的驱动版本不低于3.6.0。二、JDBC 高效写入快速上手1. 两种启用方式对于 JDBC 连接器启用高效写入有两种方式连接级启用在连接属性中设置TSDBDriver.PROPERTY_KEY_ASYNC_WRITE为stmt或在 JDBC URL 中追加asyncWrite stmt。启用后该连接上创建的所有PreparedStatement对象都将使用高效写入模式。语句级启用在使用参数绑定创建PreparedStatement时将 SQL 中的INSERT INTO改为ASYNC_INSERT INTO即可仅为该参数绑定对象启用高效写入。按 Java 连接器参考文档 的说明目前asyncWrite仅支持stmt方式且仅在 WebSocket 连接下生效。2. 如何确认写入成功客户端应用按 JDBC 标准接口的流程操作用addBatch逐条添加记录用executeBatch提交已添加的记录。在高效写入模式下executeBatch的作用是将记录提交到连接器内部的队列executeUpdate用于同步获取自上次调用以来成功写入的行数若数据写入失败此时调用executeUpdate会抛出异常可在此捕获处理。3. 核心配置参数高效写入相关的关键参数如下默认值取自 Java 连接器参考文档属性常量URL 参数名含义默认值生效条件PROPERTY_KEY_ASYNC_WRITEasyncWrite高效写入模式开关目前仅支持stmt空不启用WebSocket 连接PROPERTY_KEY_BACKEND_WRITE_THREAD_NUMbackendWriteThreadNum高效写入模式的后台写入线程数10WebSocket 连接PROPERTY_KEY_BATCH_SIZE_BY_ROWbatchSizeByRow单次写入的批大小单位为行1000WebSocket 连接PROPERTY_KEY_CACHE_SIZE_BY_ROWcacheSizeByRow缓存大小单位为行每个后台写入线程对应队列的容量10000WebSocket 连接PROPERTY_KEY_ENABLE_AUTO_RECONNECTenableAutoReconnect是否启用自动重连true/falsefalse建议高效写入模式开启WebSocket 连接PROPERTY_KEY_RECONNECT_INTERVAL_MSreconnectIntervalMs自动重连的重试间隔毫秒2000仅在enableAutoReconnecttrue时PROPERTY_KEY_RECONNECT_RETRY_COUNTreconnectRetryCount自动重连的重试次数3仅在enableAutoReconnecttrue时此外高效写入模式下还会用到以下辅助参数PROPERTY_KEY_MESSAGE_WAIT_TIMEOUTmessageWaitTimeout等待服务端处理写请求的超时时间毫秒默认 60000仅 WebSocket 连接生效PROPERTY_KEY_COPY_DATAcopyData调用addBatch后修改二进制数据时是否拷贝数据默认falsePROPERTY_KEY_STRICT_CHECKstrictCheck是否校验子表名长度及变长数据类型长度默认falsePROPERTY_KEY_VARCHAR_AS_STRINGvarcharAsString是否将 VARCHAR/BINARY 类型映射为 Java String默认false。在资源充足的前提下写入能力与写入线程数的配置近似线性相关因此可通过backendWriteThreadNum与batchSizeByRow、cacheSizeByRow的组合灵活调节吞吐。4. 最小可运行示例仓库中的 WSHighVolumeDemo.java 是官方提供的极简高效写入示例。下面提取其核心骨架已略去ANCHOR标记与完整注释完整代码请直接阅读源文件public class WSHighVolumeDemo { private static final String HOST 127.0.0.1; private static final int port 6041; private static final int NUM_OF_SUB_TABLE 10000; private static final int NUM_OF_ROW 10; public static void main(String[] args) throws SQLException { String url jdbc:TAOS-WS:// HOST : port /?userrootpasswordtaosdata; Properties properties new Properties(); // 开启高效写入模式 properties.setProperty(TSDBDriver.PROPERTY_KEY_ASYNC_WRITE, stmt); // 单次写请求最大批的行数 properties.setProperty(TSDBDriver.PROPERTY_KEY_BATCH_SIZE_BY_ROW, 10000); // 队列每个后台写线程最大缓存行数 properties.setProperty(TSDBDriver.PROPERTY_KEY_CACHE_SIZE_BY_ROW, 100000); // 后台写入线程数 properties.setProperty(TSDBDriver.PROPERTY_KEY_BACKEND_WRITE_THREAD_NUM, 5); // 连接断开后自动重连 properties.setProperty(TSDBDriver.PROPERTY_KEY_ENABLE_AUTO_RECONNECT, true); // 等待服务端处理写请求的超时时间毫秒 properties.setProperty(TSDBDriver.PROPERTY_KEY_MESSAGE_WAIT_TIMEOUT, 5000); // addBatch 之后修改二进制数据时是否拷贝 properties.setProperty(TSDBDriver.PROPERTY_KEY_COPY_DATA, false); // 是否校验子表名长度与变长类型长度 properties.setProperty(TSDBDriver.PROPERTY_KEY_STRICT_CHECK, false); properties.setProperty(TSDBDriver.PROPERTY_KEY_VARCHAR_AS_STRING, true); try (Connection conn DriverManager.getConnection(url, properties)) { init(conn); // 若确认子表已存在可不在 SQL 中绑定 tag 列以提升性能 String sql INSERT INTO power.meters (tbname, groupid, location, ts, current, voltage, phase) VALUES (?,?,?,?,?,?,?); try (PreparedStatement pstmt conn.prepareStatement(sql)) { long current System.currentTimeMillis(); int rows 0; for (int j 0; j NUM_OF_ROW; j) { for (int i 1; i NUM_OF_SUB_TABLE; i) { pstmt.setString(1, d_ i); pstmt.setInt(2, i); pstmt.setString(3, location_ i); pstmt.setTimestamp(4, new Timestamp(current j)); pstmt.setFloat(5, random.nextFloat() * 30); pstmt.setInt(6, random.nextInt(300)); pstmt.setFloat(7, random.nextFloat()); // 当后台缓存队列满时此方法会阻塞 pstmt.addBatch(); rows; } pstmt.executeBatch(); if (rows % 50000 0) { // 高效写入模式下 executeUpdate 的语义同步获取 // 上次调用到本次调用之间成功写入的行数 int affectedRows pstmt.executeUpdate(); Assert.equals(50000, affectedRows); } } } } catch (Exception ex) { // 写入异常在此捕获生产环境请使用日志框架 ex.printStackTrace(); throw ex; } } private static void init(Connection conn) throws SQLException { try (Statement stmt conn.createStatement()) { stmt.execute(CREATE DATABASE IF NOT EXISTS power); stmt.execute(USE power); stmt.execute(CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS (groupId INT, location BINARY(24))); } } }示例要点通过jdbc:TAOS-WS://前缀走 WebSocket 连接默认端口 6041即 taosAdapter 的服务端口先创建超级表meters写入时子表由连接器根据tbname自动创建addBatch在后台队列满时阻塞天然形成背压避免内存无限增长executeBatch负责把记录提交进队列executeUpdate负责同步确认写入行数并捕获异常。三、百万级吞吐实战highvolume 示例程序官方在 docs/examples/JDBC/highvolume 目录提供了一个完整的 Maven 工程用于演示真实场景下的高效写入其构建产物为可直接运行的highVolume.jar。1. 场景设计示例程序的场景设计包含三个要点也是设计高吞吐写入管道的通用思路持续读取数据源客户端程序持续从其他数据源读取数据。示例中用模拟数据生成来模仿数据源读取同时提供了从 Kafka 拉取数据并写入 TDengine 的对照实现。多线程读取、子表不重叠为提高读取速度采用多线程读取为避免乱序问题多个读取线程负责的子表集合互不重叠。读写线程解耦为匹配每个读取线程的速度后台启动一组写入线程每个写入线程拥有一个独占的固定大小消息队列。2. 类职责一览类名功能说明FastWriteExample主程序命令行参数解析、线程池创建、等待任务完成WorkTask从模拟数据源读取数据并用 JDBC 标准接口写入MockDataSource模拟生成指定数量meters子表的数据DataBaseMonitor/StatTask统计写入速度每 10 秒向控制台打印当前写入速度CreateSubTableTask在指定范围内批量创建子表供主程序调用Meters单条meters记录的序列化/反序列化用于 Kafka 消息收发ProducerTask向 Kafka 发送消息的生产者ConsumerTask从 Kafka 接收消息、用 JDBC 高效写入接口写 TDengine并按进度提交 offsetUtil基础工具创建连接、创建数据库、创建 Kafka topic、计数等各类的完整源码分别位于 FastWriteExample.java、WorkTask.java、MockDataSource.java、CreateSubTableTask.java、Meters.java、ProducerTask.java、ConsumerTask.java、StatTask.java 与 Util.java。3. 主程序命令行参数FastWriteExample使用 commons-cli 解析命令行参数参数定义见 FastWriteExample.java-b,--batchSizeByRow arg 指定高效写入的 batchSizeByRow默认 1000 -c,--cacheSizeByRow arg 指定高效写入的 cacheSizeByRow默认 10000 -d,--dbName arg 指定数据库名默认 test --help 打印帮助信息 -K,--useKafka 启用 Kafka 模式创建生产者发消息、消费者收消息写入 TDengine 否则使用工作线程订阅模拟数据写入 -r,--readThreadCount arg 指定工作线程数默认 5Kafka 模式下同时决定生产者/消费者线程数 -R,--rowsPerSubTable arg 指定每个子表写入的行数默认 100 -s,--subTableNum arg 指定子表总数默认 1000000 -w,--writeThreadPerReadThread arg 指定每个工作线程对应的写入线程数默认 5从源码可见FastWriteExample.java主程序会对各数值参数做 0的校验随后调用Util.prepareDatabase建库、createSubTables按readThreadCount均分范围并行建表最后根据是否带-K分别进入 Kafka 模式或模拟数据模式。模拟数据模式下WorkTask通过Util.getConnection(batchSizeByRow, cacheSizeByRow, writeThreadCount)建立启用了asyncWritestmt的连接见 Util.java并每batchSizeByRow行执行一次executeBatch、每10 * batchSizeByRow行执行一次executeUpdate校验见 WorkTask.java。4. 环境变量配置JDBC URL通过环境变量配置例如export TDENGINE_JDBC_URLjdbc:TAOS-WS://localhost:6041?userrootpasswordtaosdatavarcharAsStringtrue若未设置Util.getConnection 会回退到默认值jdbc:TAOS-WS://localhost:6041/?userrootpasswordtaosdatavarcharAsStringtrue本机默认部署的 TDengine。Kafka 集群地址通过环境变量配置例如export KAFKA_BOOTSTRAP_SERVERSlocalhost:9092同样地未设置时会回退到localhost:9092见 Util.java。5. 运行方式在本地 IDE 中运行获取仓库代码git clone后定位到docs/examples/JDBC/highvolume目录用 IDE 打开TDengine/docs/examples/JDBC/highvolume目录在运行环境配置TDENGINE_JDBC_URL若全局已配置可跳过运行 Kafka 示例需再设置KAFKA_BOOTSTRAP_SERVERS指定命令行参数例如-r 3 -w 3 -b 100 -c 1000 -s 1000 -R 100运行主类com.taos.example.highvolume.FastWriteExample。在远程服务器上运行# 1. 打包在 highvolume 目录下 mvn package # 2. 拷贝到服务器 scp -r .\target\highVolume.jar userhost:~/dest-path # 3. 配置环境变量编辑 ~/.bash_profile 或 ~/.bashrc export TDENGINE_JDBC_URLjdbc:TAOS-WS://localhost:6041?userrootpasswordtaosdatavarcharAsStringtrue # 如需 Kafka 订阅模式追加 export KAFKA_BOOTSTRAP_SERVERSlocalhost:9092 # 4. 启动追加 -K 进入 Kafka 订阅模式 java -jar highVolume.jar -r 5 -w 5 -b 10000 -c 100000 -s 1000000 -R 1000注意程序不会自动退出。当在当前配置下达到稳定写入速度后按CTRL C终止测试程序即可主程序注册了ShutdownHook调用forceStopAll优雅关闭各任务与线程池见 FastWriteExample.java。6. 运行日志解读以下是官方在 40 核 CPU、256GB 内存、SSD 存储的机器上的一次真实运行日志命令java -jar highVolume.jar -r 2 -w 10 -b 10000 -c 100000 -s 1000000 -R 100[INFO ] 2025-03-24 18:03:17.980 ... main readThreadCount2, writeThreadPerReadThread10 batchSizeByRow10000 cacheSizeByRow100000, subTableNum1000000, rowsPerSubTable100 [INFO ] 2025-03-24 18:03:34.499 ... main create database end. [INFO ] 2025-03-24 18:03:55.778 ... main create sub tables end. [INFO ] 2025-03-24 18:03:55.781 ... WorkTask run 41 FW-work-thread-2 started [INFO ] 2025-03-24 18:03:55.781 ... WorkTask run 41 FW-work-thread-1 started [INFO ] 2025-03-24 18:04:06.580 ... StatTask run 36 pool-1-thread-1 numberOfTable1000000 count12235906 speed1223590 [INFO ] 2025-03-24 18:04:17.531 ... StatTask run 36 pool-1-thread-1 numberOfTable1000000 count31185614 speed1894970 [INFO ] 2025-03-24 18:04:28.490 ... StatTask run 36 pool-1-thread-1 numberOfTable1000000 count51464904 speed2027929 [INFO ] 2025-03-24 18:04:40.851 ... StatTask run 36 pool-1-thread-1 numberOfTable1000000 count71498113 speed2003320 [INFO ] 2025-03-24 18:04:51.948 ... StatTask run 36 pool-1-thread-1 numberOfTable1000000 count91242103 speed1974399该日志展示了建库约 16 秒、批量建 100 万张子表约 21 秒随后StatTask每约 11 秒打印一次累计写入行数与瞬时速度speed行/秒。示例中该配置下写入速度稳定在200 万行/秒左右。需要说明这是特定硬件与配置下的示例结果实际吞吐受机器、网络、库表结构及参数影响请以自身环境的实测为准。7. Kafka 订阅写入模式启用-K后主程序进入 Kafka 模式FastWriteExample.java按readThreadCount创建对应数量的ProducerTask生产者线程与ConsumerTask消费者线程Util.createKafkaTopic创建 topictest-meters-topic分区数固定为 5见 Util.javaProducerTask把MockDataSource生成的数据发到各分区其哈希方式与 JDBC 高效写入不同这样分区与写入队列的对应关系不固定用于模拟更真实的乱序场景ConsumerTask消费消息并写入 TDengine它设置了enable.auto.commitfalse每写入10 * batchSizeByRow行调用一次executeUpdate校验成功后commitSync()提交 offset见 ConsumerTask.java实现按写入进度提交偏移的精确一次语义基础。工程依赖方面pom.xml 使用taos-jdbcdriver 3.8.3、kafka-clients 3.9.2、commons-cli 1.9.0Java 17 编译并通过maven-assembly-plugin打带依赖的 fat jar。四、影响写入性能的关键因素1. 客户端应用视角批量写入批量越大效率通常越高超过某一阈值后收益递减。用 SQL 写 TDengine 时应尽量在单条 SQL 中携带更多数据。SQL 默认最大长度约为4 MB4194304 字符可通过客户端参数maxSQLLength配置最大可到 64 MB。多线程写入在系统资源出现瓶颈前增加写入线程数可提升吞吐超过阈值后可能因服务端处理能力受限而下降。建议每个写入线程使用独立连接减少连接资源争用。写入局部性数据在不同表子表间的分布即待写数据的局部性。每个批次写单一表子表比写多张表子表更高效。预建表预先建表可省去表存在性检查且写入时可省略 tag 列数据从而提升写入性能。写入方式选择参数绑定写入比原始 SQL 写入更高效省去 SQL 解析不带自动建表的 SQL 写入比带自动建表的更高效后者需要频繁检查表是否存在SQL 写入比 schemaless 写入更高效schemaless 启用自动建表和动态 schema 变更。保序写入同一子表的数据必须按时间戳升序提交。乱序数据会在服务端引发额外排序操作影响写入性能。开启压缩当网络带宽成为瓶颈、或数据存在明显重复时开启压缩可有效提升整体性能。客户端应用应综合利用上述因素例如选择参数绑定、预建子表、每批写单一子表并通过实测配置批大小与并发线程数以获得当前系统的最优写入速度。2. 数据源视角客户端通常需要先从数据源读取数据再写入 TDengine。从数据源角度看以下情况需要在读写线程之间加队列多数据源、单源速度慢单个数据源的产生速度远低于单线程写入速度但总体数据量较大。此时队列的作用是聚合多源数据增大单次写入的数据量。单源速度极快单个数据源的产生速度远高于单线程写入速度。此时队列的作用是提高写入并发度。单表数据分散同一张表的数据分散在多个数据源中。此时队列的作用是提前聚合同一张表的数据提升写入时数据的相邻性。如果写入应用的数据源是 Kafka、且应用本身是 Kafka 消费者可直接利用 Kafka 特性将同一张表的数据写入同一 Topic 的同一 Partition提高数据相邻性订阅多个 Topic 聚合数据增加 Consumer 线程数提高写入并发增大每次拉取的数据量上限如MAX_POLL_RECORDS_CONFIG、FETCH_MAX_BYTES_CONFIG参见 ConsumerTask.java以增大单次写入的数据量。3. 服务器配置视角建库选项中的关键性能参数vgroups建库时按磁盘数量、磁盘 I/O 能力与处理器能力合理设置 vgroups 数量。vgroups 过少会浪费性能过多则带来不必要的资源争用。建议单 vgroup 内表数控制在 100 万以内硬件资源充足时控制在 1 万以内效果更佳。示例程序的Util.prepareDatabase即使用CREATE DATABASE ... vgroups 20见 Util.java。buffervnode 分配的写入内存大小默认 256 MB。当 vnode 实际写入数据达到 buffer 的约 1/3 时触发落盘。适当调大可以缓存更多数据批量刷盘提升写入效率但过大会在系统崩溃时延长恢复时间。cachemodel控制是否在内存中缓存子表最新数据。开启会影响写入性能写入时要更新每张表的last_row与每列的last值可通过把选项从both改为last_row或last_value来降低影响。stt_trigger控制 TSDB 数据刷盘策略与触发后台文件合并的文件数。企业版默认 2开源版只能配置为 1。stt_trigger 1适合表少、写入频率高的场景stt_trigger 1更适合表多、写入频率低的场景。其他建库参数可参考 数据库管理。taosd 配置中的关键参数compressMsgSize网络带宽成为瓶颈时开启 RPC 消息压缩可提升性能。numOfCommitThreads服务端后台刷盘线程数默认 4。线程数并非越多越好过多会引起磁盘写争用多磁盘服务器可考虑增大该参数以利用并发 I/O 能力。日志级别debugFlag等参数控制日志输出级别。日志级别越高输出压力越大影响写入性能建议使用默认配置。其他服务端参数可参考 taosd 服务配置。五、高效写入的实现原理从上述影响写入性能的因素可以看出开发高性能写入程序需要多线程编程与数据分片知识存在技术门槛。为降低用户开发成本TDengine 连接器提供高效写入特性让用户不必处理底层线程管理与数据分片逻辑即可利用 TDengine 强大的写入能力。下图是连接器高效写入功能的实现原理示意图结合架构图与源码可以梳理出整体数据流应用Application通过写接口把请求交给应用线程Thread A→ 连接器按子表将数据分片投递到各自专属队列Queue 1/2/3→ 后台写入线程组Write Thread 1/2/3从对应队列批量取出数据并打包发送 → 数据最终写入 TDengine 集群TDengine Cluster。1. 设计原则自动创建线程与队列连接器根据配置参数动态创建独立的写入线程与对应写入队列每个队列绑定一个子表形成子表数据 → 专属队列 → 独立线程的处理链。数据分片与批量触发应用写入数据时连接器自动按子表分片并缓存在对应队列满足以下任一条件即触发批量发送队列数据量达到预设阈值即batchSizeByRow/cacheSizeByRow相关机制达到预设等待超时避免延迟过大。该机制通过减少网络请求次数显著提升写入吞吐。2. 功能优势单线程高性能写入应用层只需单线程调用写接口连接器底层自动处理多线程并发可获得接近传统客户端多线程写入的性能同时屏蔽所有底层线程管理复杂性。从源码看WorkTask以阻塞方式调用addBatch——当对应高效写入后台队列满时写操作会阻塞见 WorkTask.java 及 WSHighVolumeDemo 注释这构成了天然的反压保护。可靠性增强机制同步校验接口提供同步方法executeUpdate确保提交数据的写入成功状态可追踪连接自愈支持断线后自动重连配合可配置的超时重试策略重试次数与间隔避免数据丢失错误隔离单个队列或线程的异常不影响其他子表的数据写入提升系统容错性。注意连接器的高效写入特性仅支持超级表supertable写入不支持普通表。从 WorkTask.java 与 ConsumerTask.java 可见示例均以dbName.meters这一超级表为目标若实际场景涉及多张超级表只需修改写任务中自动建表的代码即可。六、总结与实践建议TDengine 连接器的高效写入特性把多线程 队列 分片 批量这套高吞吐写入工程实践内置到驱动层开发者只需通过asyncWritestmt开启、配置batchSizeByRow、cacheSizeByRow、backendWriteThreadNum三个核心参数即可在单线程编程模型下获得接近传统多线程方案的写入性能。落地高吞吐写入管道时建议按以下顺序排查与调优客户端优先使用参数绑定 预建子表 每批写单一子表用executeUpdate周期性同步校验写入行数保证写入可追踪架构读写线程之间用队列解耦可用高效写入自带的后台队列或数据源侧显式加队列/Kafka同一子表的数据尽量保序、聚合服务器按磁盘与 CPU 能力合理配置vgroups、buffer、stt_trigger、numOfCommitThreads带宽受限时开启压缩验证参照本文日志解读方法用StatTask之类的统计任务观察稳定写入速度再逐步调大并发与批大小直至系统资源成为瓶颈。进一步阅读完整示例代码见 highvolume 工程最小示例见 WSHighVolumeDemo.java全部连接器配置参数见 Java 连接器参考。【免费下载链接】tdengineTDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.项目地址: https://gitcode.com/taosdata/tdengine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
