feat(processor): 实现批量写库容错机制并添加失败统计

添加数据库批量写入失败处理逻辑，当批量写入失败时自动切换为逐条写入记录失败数据并统计失败数量，同时更新相关测试和统计模块符合新增的批量写库容错需求规范
2026-01-20 08:22:55 +08:00
parent 41301f9ce5
commit b90faf4aa4
5 changed files with 111 additions and 44 deletions
--- a/src/db/databaseManager.js
+++ b/src/db/databaseManager.js
@@ -425,46 +425,51 @@ class DatabaseManager {
      'extra',
    ];

+    const toRowValues = (e) => [
+      e.ts_ms,
+      e.hotel_id,
+      e.room_id,
+      e.device_id,
+      e.ip,
+      e.power_state,
+      e.guest_type,
+      e.cardless_state,
+      e.service_mask,
+      e.pms_state,
+      e.carbon_state,
+      e.device_count,
+      e.comm_seq,
+      Array.isArray(e.elec_address) ? e.elec_address : null,
+      Array.isArray(e.air_address) ? e.air_address : null,
+      Array.isArray(e.voltage) ? e.voltage : null,
+      Array.isArray(e.ampere) ? e.ampere : null,
+      Array.isArray(e.power) ? e.power : null,
+      Array.isArray(e.phase) ? e.phase : null,
+      Array.isArray(e.energy) ? e.energy : null,
+      Array.isArray(e.sum_energy) ? e.sum_energy : null,
+      Array.isArray(e.state) ? e.state : null,
+      Array.isArray(e.model) ? e.model : null,
+      Array.isArray(e.speed) ? e.speed : null,
+      Array.isArray(e.set_temp) ? e.set_temp : null,
+      Array.isArray(e.now_temp) ? e.now_temp : null,
+      Array.isArray(e.solenoid_valve) ? e.solenoid_valve : null,
+      e.extra ?? null,
+    ];
+
    const values = [];
    const placeholders = events
      .map((e, rowIndex) => {
        const base = rowIndex * columns.length;
-        values.push(
-          e.ts_ms,
-          e.hotel_id,
-          e.room_id,
-          e.device_id,
-          e.ip,
-          e.power_state,
-          e.guest_type,
-          e.cardless_state,
-          e.service_mask,
-          e.pms_state,
-          e.carbon_state,
-          e.device_count,
-          e.comm_seq,
-          Array.isArray(e.elec_address) ? e.elec_address : null,
-          Array.isArray(e.air_address) ? e.air_address : null,
-          Array.isArray(e.voltage) ? e.voltage : null,
-          Array.isArray(e.ampere) ? e.ampere : null,
-          Array.isArray(e.power) ? e.power : null,
-          Array.isArray(e.phase) ? e.phase : null,
-          Array.isArray(e.energy) ? e.energy : null,
-          Array.isArray(e.sum_energy) ? e.sum_energy : null,
-          Array.isArray(e.state) ? e.state : null,
-          Array.isArray(e.model) ? e.model : null,
-          Array.isArray(e.speed) ? e.speed : null,
-          Array.isArray(e.set_temp) ? e.set_temp : null,
-          Array.isArray(e.now_temp) ? e.now_temp : null,
-          Array.isArray(e.solenoid_valve) ? e.solenoid_valve : null,
-          e.extra ?? null
-        );
+        values.push(...toRowValues(e));
        const row = columns.map((_, colIndex) => `$${base + colIndex + 1}`).join(', ');
        return `(${row})`;
      })
      .join(', ');

    const sql = `INSERT INTO heartbeat.heartbeat_events (${columns.join(', ')}) VALUES ${placeholders}`;
+    const singleSql = `INSERT INTO heartbeat.heartbeat_events (${columns.join(', ')}) VALUES (${columns
+      .map((_, i) => `$${i + 1}`)
+      .join(', ')})`;

    const runInsertOnce = async () => {
      const tsValues = events.map((e) => Number(e.ts_ms)).filter((n) => Number.isFinite(n));
@@ -518,7 +523,24 @@ class DatabaseManager {
      }
    }

-    throw lastError;
+    const failedRecords = [];
+    let insertedCount = 0;
+    console.error('[db] 批量写入失败，已切换为逐条写入:', lastError);
+
+    for (const event of events) {
+      try {
+        await this.pool.query(singleSql, toRowValues(event));
+        insertedCount += 1;
+      } catch (error) {
+        failedRecords.push({ error, record: event });
+      }
+    }
+
+    if (insertedCount === 0 && failedRecords.length === events.length) {
+      throw lastError;
+    }
+
+    return { insertedCount, failedRecords, batchError: lastError };
  }

  async insertHeartbeatData(data) {
--- a/src/processor/heartbeatProcessor.js
+++ b/src/processor/heartbeatProcessor.js
@@ -204,18 +204,16 @@ class HeartbeatProcessor {
      const batchMessages = this.batchMessageQueue.slice(0, batchMessageCount);

      let insertedCount = 0;
+      let failedRecords = [];
      if (typeof this.databaseManager.insertHeartbeatEvents === 'function') {
        const result = await this.databaseManager.insertHeartbeatEvents(batchData);
        insertedCount = Number(result?.insertedCount ?? result ?? 0);
+        failedRecords = Array.isArray(result?.failedRecords) ? result.failedRecords : [];
      } else {
        const result = await this.databaseManager.insertHeartbeatData(batchData);
        insertedCount = Number(result?.insertedCount ?? result ?? 0);
      }

-      if (insertedCount !== batchData.length) {
-        throw new Error(`落库结果校验失败：expect=${batchData.length} actual=${insertedCount}`);
-      }
-
      this.batchQueue.splice(0, batchEventCount);
      this.batchMessageQueue.splice(0, batchMessageCount);

@@ -223,8 +221,23 @@ class HeartbeatProcessor {
        entry.deferred.resolve({ insertedCount: entry.eventCount });
      }

-      this.stats?.incDbWritten?.(batchData.length);
-      console.log(`成功处理批次数据，共 ${batchData.length} 条`);
+      if (failedRecords.length > 0) {
+        for (const item of failedRecords) {
+          this._emitRejectedRecord({
+            errorId: 'db_write_failed',
+            error: item?.error,
+            rawData: item?.record,
+          });
+        }
+        this.stats?.incDbWriteFailed?.(failedRecords.length);
+      }
+      this.stats?.incDbWritten?.(insertedCount);
+      const failedCount = failedRecords.length;
+      if (failedCount > 0) {
+        console.log(`批次处理完成：成功 ${insertedCount} 条，失败 ${failedCount} 条`);
+      } else {
+        console.log(`成功处理批次数据，共 ${batchData.length} 条`);
+      }
      hasMore = this.batchQueue.length > 0;
    } catch (error) {
      console.error('批量处理失败:', error);
@@ -244,8 +257,11 @@ class HeartbeatProcessor {
  }

  _emitDbWriteError(error, rawData) {
-    if (!this.redis?.isEnabled?.()) return;
    const list = Array.isArray(rawData) ? rawData : rawData ? [rawData] : [];
+    if (list.length > 0) {
+      this.stats?.incDbWriteFailed?.(list.length);
+    }
+    if (!this.redis?.isEnabled?.()) return;
    for (const record of list) {
      this._emitRejectedRecord({
        errorId: 'db_write_failed',
--- a/src/stats/statsManager.js
+++ b/src/stats/statsManager.js
@@ -1,6 +1,6 @@
 class StatsCounters {
  constructor() {
-    this._minuteBuf = new SharedArrayBuffer(BigInt64Array.BYTES_PER_ELEMENT * 3);
+    this._minuteBuf = new SharedArrayBuffer(BigInt64Array.BYTES_PER_ELEMENT * 4);
    this._minute = new BigInt64Array(this._minuteBuf);
  }

@@ -22,11 +22,18 @@ class StatsCounters {
    Atomics.add(this._minute, 2, v);
  }

+  incDbWriteFailed(n = 1) {
+    const v = BigInt(Math.max(0, Number(n) || 0));
+    if (v === 0n) return;
+    Atomics.add(this._minute, 3, v);
+  }
+
  snapshotAndResetMinute() {
    const dbWritten = Atomics.exchange(this._minute, 0, 0n);
    const filtered = Atomics.exchange(this._minute, 1, 0n);
    const kafkaPulled = Atomics.exchange(this._minute, 2, 0n);
-    return { dbWritten, filtered, kafkaPulled };
+    const dbWriteFailed = Atomics.exchange(this._minute, 3, 0n);
+    return { dbWritten, filtered, kafkaPulled, dbWriteFailed };
  }
 }

@@ -50,6 +57,7 @@ class StatsReporter {
    this.stats = stats;
    this._timer = null;
    this._running = false;
+    this._lastFlushMinute = null;
  }

  start() {
@@ -68,9 +76,16 @@ class StatsReporter {

  flushOnce() {
    if (!this.redis?.isEnabled?.()) return;
-    const { dbWritten, filtered, kafkaPulled } = this.stats.snapshotAndResetMinute();
+    const now = Date.now();
+    const minuteKey = Math.floor(now / 60_000);
+    if (this._lastFlushMinute === minuteKey) {
+      return;
+    }
+    const { dbWritten, filtered, kafkaPulled, dbWriteFailed } = this.stats.snapshotAndResetMinute();
+    this._lastFlushMinute = minuteKey;
    const ts = formatTimestamp(new Date());
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} 数据库写入量: ${dbWritten}条`, metadata: { module: 'stats' } });
+    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} 数据库写入失败量: ${dbWriteFailed}条`, metadata: { module: 'stats' } });
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} 数据过滤量: ${filtered}条`, metadata: { module: 'stats' } });
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} Kafka拉取量: ${kafkaPulled}条`, metadata: { module: 'stats' } });
  }