Cloud Function随机未向PubSub发送消息且无报错的问题咨询

1. 当前实现存在错误，核心是异步执行逻辑不符合Cloud Function的运行要求 #

Cloud Function的运行时会在入口函数返回后立刻回收执行资源，你当前的代码没有将异步操作的执行状态暴露给运行时，导致发布消息的逻辑经常在执行完成前就被中断，这就是随机无日志、无消息发出的根因：

• 你使用回调形式调用conn.query，没有将其转为Promise等待执行完成
• 循环调用异步函数msgPubSub时没有收集所有Promise并等待全部执行完成
• 入口函数没有返回对应异步操作的Promise，运行时误以为逻辑已经执行结束

2. 你对PubSub触发的Cloud Function的确认逻辑理解正确，不需要手动确认消息 #

PubSub触发的Cloud Function在执行成功（无异常抛出、返回正常resolved的Promise）时会自动向PubSub返回确认，执行失败时会返回nack触发重试，这个逻辑不需要你手动实现。你当前的故障出在负责发布消息的第一个Cloud Function，和第二个消费函数的确认逻辑无关。

3. 无报错本质是执行被中断导致日志未输出，修复异步问题后再实现重发逻辑 #

首先你需要先修正异步逻辑，让所有操作完成后CF再终止，此时才能拿到真实的成功/失败状态：

第一步：修正异步逻辑 #

参考修正后的代码：

//[requirements]
const {PubSub} = require('@google-cloud/pubsub');
const pubSubClient = new PubSub('<PROJECT_ID>');
const topicName = "<TOPIC_NAME>";
// 配置发布端自带的重试策略
const topicPublisher = pubSubClient.topic(topicName, {
  retry: {
    retryCodes: [10, 1, 2, 4, 13, 8, 14], // 对应PubSub可重试的错误码
    backoffSettings: {
      initialRetryDelayMillis: 100,
      retryDelayMultiplier: 2,
      maxRetryDelayMillis: 60000,
      initialRpcTimeoutMillis: 5000,
      rpcTimeoutMultiplier: 2,
      maxRpcTimeoutMillis: 60000,
      totalTimeoutMillis: 600000,
    }
  }
})
//[requirements]

// 把回调形式的query转为Promise
function querySQL(conn, sql) {
  return new Promise((resolve, reject) => {
    conn.query(sql, (e, results) => {
      if(e) return reject(e);
      resolve(results);
    })
  })
}

//[PubSub message publish function]
async function msgPubSub(data){
  const messageBuffer = Buffer.from(data)
  try {
    const messageId = await topicPublisher.publish(messageBuffer)
    console.log("Message id: " + messageId)
    return messageId;
  } catch (error) {
    console.error(`Error while publishing message: ${error.message}，内容：${data}`)
    throw error; // 抛出错误让外层感知失败
  }
}

// 入口函数示例，需要返回Promise
exports.yourEntryFunction = async (event, context) => {
  let conn; // 你的数据库连接实例
  const sql = "你的查询语句";
  try {
    const results = await querySQL(conn, sql);
    console.log(`查询到${results.length}条数据，开始发布到PubSub`);
    // 等待所有消息发布完成
    await Promise.all(results.map(element => {
      const str = JSON.stringify(element);
      console.log(str);
      return msgPubSub(str);
    }))
    console.log("所有消息发布完成");
  } catch (e) {
    console.error("执行失败：", e);
    throw e; // 抛出错误触发Cloud Function重试
  } finally {
    if(conn) conn.end(); // 关闭数据库连接
  }
}

第二步：重发逻辑实现 #

修正异步逻辑后如果还是出现偶发发布失败，可以通过以下方式处理：

• 依赖PubSub客户端自带的重试配置，如上面代码所示，客户端会对可重试的错误自动执行退避重试
• 配置Cloud Function的重试策略，对于执行失败的第一个CF自动重试，注意要给消息加唯一业务ID实现消费幂等，避免重复执行导致重复数据
• 对于多次重试失败的消息，可以写入死信队列单独处理，避免阻塞主流程

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