如何在Firebase云函数中使用Google Cloud Task Queue并获取JSON响应

Google Cloud Tasks本身是异步队列服务，创建任务时只会返回任务的元数据（如任务名称），不会直接返回任务的执行结果。要实现类似Redis队列的"任务完成后获取结果继续流程"的需求，需要自行设计结果存储+主动查询/回调通知的机制，以下是具体实现方案：

方案一：结果存储+主动轮询 #

通过Firebase Firestore（或Realtime Database）存储任务结果，发起任务的函数定期查询存储直到获取结果。

修改任务处理函数（addNumbersFunc） #

执行计算后将结果存入Firestore，用唯一任务ID作为标识：

const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.addNumbersFunc = functions.https.onRequest(async (req, res) => {
  const payload = req.body;
  const taskId = payload.taskId;
  const answer = payload.numbers.a + payload.numbers.b;

  // 将结果写入Firestore
  await firestore.collection('taskResults').doc(taskId).set({
    answer: answer,
    status: 'completed',
    timestamp: new Date()
  });

  res.json({ answer: answer });
});

修改发起任务的函数（exampleFunction1） #

生成唯一任务ID，创建任务后轮询Firestore获取结果：

const { v4: uuidv4 } = require('uuid');
const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.exampleFunction1 = functions.https.onCall(async (data) => {
  // 生成唯一任务ID
  const taskId = uuidv4();

  const jsonPayload = JSON.stringify({
    taskId: taskId,
    numbers: { a: 1, b: 2 }
  });

  const url = `https://${location}-${project}.cloudfunctions.net/addNumbersFunc`;

  const task = {
    httpRequest: {
      httpMethod: 'POST',
      url: url,
      headers: { 'Content-Type': 'application/json' },
      body: Buffer.from(jsonPayload).toString('base64')
    },
    oidcToken: { serviceAccountEmail }
  };

  const request = { parent: parent, task: task };
  const [response] = await client.createTask(request);
  console.log(`Created task ${response.name}`);

  // 轮询获取结果（最多等待30秒，每秒查询一次）
  const maxWaitTime = 30000;
  const pollInterval = 1000;
  let elapsedTime = 0;
  let taskResult = null;

  while (elapsedTime < maxWaitTime) {
    const doc = await firestore.collection('taskResults').doc(taskId).get();
    if (doc.exists) {
      taskResult = doc.data();
      break;
    }
    await new Promise(resolve => setTimeout(resolve, pollInterval));
    elapsedTime += pollInterval;
  }

  if (!taskResult) {
    throw new functions.https.HttpsError('deadline-exceeded', '任务执行超时');
  }

  // 可选：获取结果后删除存储记录
  await firestore.collection('taskResults').doc(taskId).delete();

  return {
    taskName: response.name,
    answer: taskResult.answer
  };
});

方案二：回调通知 #

任务处理完成后主动调用预设的回调URL，将结果传递回去，适合长时间运行的任务。

修改任务处理函数 #

执行完成后调用回调URL：

const axios = require('axios');

exports.addNumbersFunc = functions.https.onRequest(async (req, res) => {
  const payload = req.body;
  const answer = payload.numbers.a + payload.numbers.b;
  const callbackUrl = payload.callbackUrl;
  const taskId = payload.taskId;

  // 调用回调URL传递结果
  try {
    await axios.post(callbackUrl, {
      taskId: taskId,
      answer: answer
    });
  } catch (err) {
    console.error('回调失败:', err);
  }

  res.json({ answer: answer });
});

新增回调函数 #

接收处理函数的结果并存储（或直接通知发起方）：

const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.taskCallback = functions.https.onRequest(async (req, res) => {
  const { taskId, answer } = req.body;
  await firestore.collection('taskResults').doc(taskId).set({
    answer: answer,
    status: 'completed'
  });
  res.status(200).send('回调已处理');
});

修改发起任务的函数 #

在Payload中加入回调URL和任务ID，之后可通过查询Firestore获取结果：

const { v4: uuidv4 } = require('uuid');
const { Firestore } = require('@google-cloud/firestore');
const firestore = new Firestore();

exports.exampleFunction1 = functions.https.onCall(async (data) => {
  const taskId = uuidv4();
  // 回调URL为新增的taskCallback函数地址
  const callbackUrl = `https://${location}-${project}.cloudfunctions.net/taskCallback`;

  const jsonPayload = JSON.stringify({
    taskId: taskId,
    numbers: { a: 1, b: 2 },
    callbackUrl: callbackUrl
  });

  const url = `https://${location}-${project}.cloudfunctions.net/addNumbersFunc`;

  const task = {
    httpRequest: {
      httpMethod: 'POST',
      url: url,
      headers: { 'Content-Type': 'application/json' },
      body: Buffer.from(jsonPayload).toString('base64')
    },
    oidcToken: { serviceAccountEmail }
  };

  const request = { parent: parent, task: task };
  const [response] = await client.createTask(request);
  console.log(`Created task ${response.name}`);

  // 此处可根据业务选择：
  // 1. 立即返回任务ID，让前端自行轮询结果
  // 2. 继续轮询Firestore直到获取结果（同方案一的轮询逻辑）
  return { taskName: response.name, taskId: taskId };
});

注意事项 #

• 限流实现：Cloud Tasks本身支持通过队列配置（并发数、速率限制）实现HTTP接口的限流，无需额外代码处理。
• 权限控制：确保只有授权的服务账号能写入taskResults集合，回调URL需验证请求来源（如通过OIDC令牌）防止恶意调用。
• 超时与重试：根据任务实际执行时间调整轮询的超时时间，Cloud Tasks默认会重试失败的任务，需避免重复写入结果。

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