使用Node.js+Google Pub/Sub+csv-parse消息超量发布及报错求助

让咱们一步步拆解你的问题，先从核心矛盾入手——大文件处理时消息爆炸和未捕获的Promise错误，本质是并发控制缺失+异步错误未处理+客户端重试逻辑触发重复发布导致的，下面是具体的修复步骤：

1. 先解决UnhandledPromiseRejection错误 #

你当前的代码中，publishToTopic返回的Promise被reject时没有任何catch处理，直接触发了全局未捕获拒绝错误。这不仅会导致程序抛出警告，还可能让错误状态无法被感知，进而引发后续逻辑异常。

修改发布逻辑，添加错误捕获：

let topicPublishResult = PubSubPublish.publishToTopic(process.env.IMPORT_CSV_ROW_PUBLISHING_TOPIC, messageObject);
topicPublishResult.then((response) => {
  rowCounter += 1;
  const messageInfo = `Row ${rowCounter} published | MessageId = ${response} | importId = ${this.data.importId} | fileId = ${this.data.fileId} | orgId = ${this.data.orgId}`;
  console.info(messageInfo);
}).catch((err) => {
  // 记录错误日志，可选：保存失败行信息以便后续重试
  console.error(`Failed to publish row ${cnt} (importId: ${this.data.importId}):`, err);
});

2. 控制并发发布数量，避免请求积压 #

处理6000条记录时，你的代码会瞬间发起6000个Publish请求，远超客户端和PubSub服务的处理能力，导致请求超时触发google-gax的自动重试——这就是消息数从6000暴增至4-5万的核心原因。

添加并发限制，比如每次最多同时发布50条：

async processFile(filename) {
  let cnt = 0;
  let index = null;
  let rowCounter = 0;
  const concurrencyLimit = 50; // 根据服务器性能调整，推荐50-100
  let activePublishPromises = [];

  const handler = async (resolve, reject) => {
    const parser = CsvParser({ delimiter: ',', })
      .on('readable', async () => {
        let row;
        this.meta.totalRows = (parser.info.records - 1);
        while (row = parser.read()) {
          if (cnt++ === 0) {
            index = row;
            continue;
          }

          // 控制并发数：达到限制时等待任意一个请求完成
          if (activePublishPromises.length >= concurrencyLimit) {
            await Promise.race(activePublishPromises);
            activePublishPromises = activePublishPromises.filter(p => !p.isCompleted);
          }

          let messageObject = {
            customFieldsMap: this.customFieldsMap,
            importAttributes: this.jc.attrs,
            importColumnData: row,
            rowCount: cnt,
            importColumnList: index,
            authToken: this.token
          };

          const publishPromise = PubSubPublish.publishToTopic(process.env.IMPORT_CSV_ROW_PUBLISHING_TOPIC, messageObject)
            .then((response) => {
              rowCounter += 1;
              console.info(`Row ${rowCounter} published | MessageId = ${response} | importId = ${this.data.importId}`);
            })
            .catch((err) => {
              console.error(`Failed to publish row ${cnt} (importId: ${this.data.importId}):`, err);
            })
            .finally(() => {
              publishPromise.isCompleted = true;
            });

          publishPromise.isCompleted = false;
          activePublishPromises.push(publishPromise);
        }
      })
      .on('end', async () => {
        // 等待所有剩余的Publish请求完成
        await Promise.all(activePublishPromises);
        console.log("File consumed!");
        resolve(this.setStatus("queued"));
      })
      .on('error', reject);
    fs.createReadStream(filename).pipe(parser);
  };
  await new Promise(handler);
}

3. 调整PubSub客户端的重试与批处理配置 #

你的错误提示Retry total timeout exceeded说明Publish请求在多次重试后仍未成功，客户端的默认重试逻辑会重复发送消息。优化批处理和重试参数，减少不必要的重试：

// 发布模块代码修改
const { PubSub } = require('@google-cloud/pubsub');
const pubsub = new PubSub({ projectId: process.env.PROJECT_ID });
module.exports = {
  publishToTopic: function(topicName, data) {
    return pubsub.topic(topicName, {
      batching: {
        maxMessages: 1000, // 增大批量大小，减少请求次数
        maxMilliseconds: 2000, // 缩短批量等待时间，避免超时
      },
      retry: {
        totalTimeout: 60000, // 延长总超时时间到60秒
        maxRetryDelay: 10000, // 最大重试间隔10秒
        retryDelayMultiplier: 1.2, // 减缓重试间隔增长速度
        // 只对可重试的错误码进行重试
        retryCodes: [10, 1, 2, 4, 5]
      }
    }).publish(Buffer.from(JSON.stringify(data)));
  },
};

4. 实现订阅端幂等性（关键兜底措施） #

即使做了上述调整，网络波动仍可能导致重复消息。在订阅端处理消息时，用rowCount + importId作为唯一标识，处理前先检查这条记录是否已经被处理过，避免重复调用第三方API和插入数据库。

额外优化建议 #

• 升级PubSub依赖：你当前使用的@google-cloud/pubsub@1.5.0版本较旧，新版本修复了大量重试和批处理的bug，建议升级到最新稳定版（如4.x）。
• 监控PubSub指标：在Google Cloud Console查看publish request latency、message duplication rate等指标，帮助定位性能瓶颈。
• 拆分大文件：对于超大规模的CSV，可先拆分为多个小文件再处理，降低单批次压力。

内容的提问来源于stack exchange，提问作者SaharshJ