Node.js连接TDengine连接池与批量写入优化实践

脚本专家

在Node.js中通过@tdengine/websocket连接TDengine时，生产环境不能只停留在单连接查询。原文围绕WebSocket连接器给出了连接池、健康检查、预编译批量写入、并行写入、无模式写入和参数绑定等可落地模式。示例配置使用ws://localhost:6041、root、taosdata、test和5000毫秒超时，适用于需要连接复用和高频写入的Node.js服务。

一、连接池设计
原文实现TDengineConnectionPool类，构造参数为config和poolSize，默认池大小10。内部维护connections、availableConnections、waitingQueue和initialized。createConnection用taos.WSConfig创建配置，依次setUser、setPwd、setDb、setTimeOut，再调用taos.sqlConnect。getConnection优先从availableConnections弹出连接，没有可用连接时把resolve放入waitingQueue等待；releaseConnection若有等待者就唤醒，否则归还到可用列表。executeQuery和execute都在finally中释放连接，避免异常时连接泄漏。

2. const taos = require('@tdengine/websocket');
3. class TDengineConnectionPool {
4.   constructor(config, poolSize = 10) {
5.     this.config = config;
6.     this.poolSize = poolSize;
7.     this.connections = [];
8.     this.availableConnections = [];
9.     this.waitingQueue = [];
10.     this.initialized = false;
11.   }
12.   async createConnection() {
13.     const wsConfig = new taos.WSConfig(this.config.url);
14.     wsConfig.setUser(this.config.user);
15.     wsConfig.setPwd(this.config.password);
16.     wsConfig.setDb(this.config.database);
17.     wsConfig.setTimeOut(this.config.timeout || 5000);
18.     return await taos.sqlConnect(wsConfig);
19.   }
20.   async getConnection() {
21.     if (!this.initialized) await this.initialize();
22.     if (this.availableConnections.length > 0) return this.availableConnections.pop();
23.     return new Promise((resolve) => { this.waitingQueue.push(resolve); });
24.   }
25.   releaseConnection(connection) {
26.     if (this.waitingQueue.length > 0) {
27.       const resolve = this.waitingQueue.shift();
28.       resolve(connection);
29.     } else {
30.       this.availableConnections.push(connection);
31.     }
32.   }
33.   async executeQuery(sql, reqId) {
34.     const conn = await this.getConnection();
35.     try { return await conn.query(sql, reqId); }
36.     finally { this.releaseConnection(conn); }
37.   }
38. }

复制代码

调用侧先initialize，再并发执行查询，最后在finally中close。原文示例用20个并发查询验证池化效果。需要注意：该实现等待队列没有超时参数，若连接长期不释放，等待方会一直挂起；排查时应先看初始化日志、可用连接数量和执行路径是否都走了releaseConnection。

二、连接健康检查
HealthCheckPool继承连接池，新增checkInterval，默认30000毫秒。initialize后启动setInterval，performHealthCheck对每个连接执行SELECT SERVER_VERSION()，失败时记录连接索引和连接对象，再调用replaceConnection：先关闭旧连接，再createConnection，并同步更新connections和availableConnections中的旧引用。close时先clearInterval再调用父类close，防止定时器残留。健康检查失败通常与连接失效、服务端不可达或认证配置变化有关，日志中的连接索引是定位入口。

三、预编译批量写入
原文用STMT处理meters超级表。表结构为ts TIMESTAMP、current FLOAT、voltage INT、phase FLOAT，标签为location NCHAR(64)、groupId INT。先stmtInit，再prepare语句，表名和标签通过setTableName、setTags传入，数据列通过bindParam按数组绑定，最后batch和exec。示例batchSize为1000，numBatches为10，共10000条；时间戳用BigInt构造，最终按总条数除以耗时计算每秒记录数。

2. stmt = await conn.stmtInit();
3. const prepareSql = 'INSERT INTO ? USING meters TAGS(?, ?) VALUES(?, ?, ?, ?)';
4. await stmt.prepare(prepareSql);
5. const batchSize = 1000;
6. const numBatches = 10;
7. for (let batch = 0; batch < numBatches; batch++) {
8.   await stmt.setTableName(`d${batch % 10}`);
9.   await stmt.setTags([`location_${batch % 10}`, batch % 10]);
10.   const timestamps = [];
11.   const currents = [];
12.   const voltages = [];
13.   const phases = [];
14.   const baseTimestamp = BigInt(Date.now() - (batchSize * 1000));
15.   for (let i = 0; i < batchSize; i++) {
16.     timestamps.push(baseTimestamp + BigInt(i * 1000));
17.     currents.push(Math.random() * 10);
18.     voltages.push(Math.floor(Math.random() * 220) + 200);
19.     phases.push(Math.random() * 360);
20.   }
21.   await stmt.bindParam([timestamps, currents, voltages, phases]);
22.   await stmt.batch();
23. }
24. await stmt.exec();

复制代码

错误排查重点：prepare的占位符数量要与bindParam数组对应；表名和标签先设置再绑定数据；BigInt时间戳不要混入普通Number；finally中先关闭stmt再关闭conn。

四、并行批量写入
为了提升吞吐量，原文用numWorkers为5、recordsPerWorker为100000，总计500000条。每个worker独立建立连接和STMT，表名为d_worker${workerId}_${batch % 10}，标签为worker_${workerId}和workerId，batchSize仍为1000，numBatches用Math.ceil计算。所有worker放入Promise.all，任一worker抛错都会使整体失败；因此每个worker的catch要记录workerId，finally关闭stmt和conn。总体性能同样按总记录数除以总耗时计算。

五、无模式写入与参数绑定
无模式写入示例生成InfluxDB行协议，时间戳换算为纳秒，batchSize为1000，调用conn.schemalessInsert时传入taos.SchemalessProto.InfluxDBLineProtocol、taos.Precision.NANO_SECONDS和ttl为0。行协议中包含meters、location=California、groupId=1以及current、voltage、phase字段。

2. await conn.schemalessInsert(
3.   lines,
4.   taos.SchemalessProto.InfluxDBLineProtocol,
5.   taos.Precision.NANO_SECONDS,
6.   0
7. );

复制代码

高级参数绑定示例建sensor_data超级表，字段包括ts TIMESTAMP、temperature FLOAT、humidity INT、pressure DOUBLE、status BOOL、device_name NCHAR(50)、raw_data BINARY(100)，标签为sensor_id NCHAR(50)、location NCHAR(100)、install_date TIMESTAMP。setTableName设为sensor_001，setTags传入字符串和BigInt时间戳，bindParam按列数组绑定多行，再batch和exec。

2. await stmt.setTableName('sensor_001');
3. await stmt.setTags([
4.   'SENSOR-001',
5.   'Building A, Floor 3',
6.   BigInt(Date.now() - 86400000 * 30)
7. ]);
8. await stmt.bindParam([
9.   timestamps, temperatures, humidities, pressures,
10.   statuses, deviceNames, rawData
11. ]);
12. await stmt.batch();
13. await stmt.exec();

复制代码

六、TMQ与生产化主题的边界
原文目录还列出TMQ消费者、多消费组、错误重试、流式结果、缓存、监控指标、配置管理、优雅关闭和日志等主题，但正文在TDengineTMQConsumer.initialize处截断，只确认该类维护config、consumer、running、messageHandlers等成员。未给出的消费循环、多消费组和重试实现不应臆造，落地时应以连接器实际API为准补齐。

总体来看，原文给出的可复用路径是：用连接池复用WebSocket连接，用健康检查替换失效连接，用STMT和1000条批量降低写入开销，用多worker并行提升吞吐，用无模式写入承接InfluxDB行协议数据。排查时按连接配置、连接池释放、健康检查日志、STMT占位符与绑定类型、批量执行和关闭顺序逐层定位，能覆盖大部分Node.js连接TDengine的常见问题。