Cloudflare Worker中外部服务调用的批量处理问题

你的当前实现存在核心问题：Cloudflare Worker是无状态运行环境，全局变量batch是实例级的——每个Worker实例会独立维护自己的batch，而Cloudflare会根据请求量动态创建/销毁实例，不同请求大概率跑在不同实例上。这就导致部署后，你的批次永远凑不到100条，自然不会触发Kinesis Firehose的发送逻辑；而本地运行是单实例，所以能正常工作。

修正方案：用Cloudflare KV实现持久化批次 #

Cloudflare KV是低成本的持久化存储（免费额度足够小流量场景），可以跨Worker实例共享批次数据，完美解决你的问题。同时还能添加超时逻辑，避免数据因凑不到批次大小一直积压。

修改后的代码示例 #

import { createUrl, getFinalUrl, getUserId, isValidUrl } from "./lib/url";
import { sendToFirehose } from "./lib/sendToFirehose";

export interface Env {
  AWS_ACCESS_KEY_ID: string;
  AWS_SECRET_ACCESS_KEY: string;
  LOG_BATCH_KV: KVNamespace; // 需在Cloudflare控制台创建对应KV命名空间
}

const BATCH_SIZE = 100;
const MAX_BATCH_AGE_MINUTES = 5; // 最长积压5分钟，到点就发送

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const logData = {
      date: new Date().toISOString(), // 存字符串避免Date对象序列化问题
      url: request.url,  
    };

    // 从KV读取当前批次及创建时间
    let batchRecord = await env.LOG_BATCH_KV.get("log-batch", "json") || {
      batch: [],
      createdAt: new Date().toISOString()
    };
    const { batch, createdAt } = batchRecord;

    // 添加新日志
    batch.push(logData);

    const now = new Date();
    const batchAge = (now.getTime() - new Date(createdAt).getTime()) / (1000 * 60);
    const shouldSend = batch.length >= BATCH_SIZE || batchAge >= MAX_BATCH_AGE_MINUTES;

    if (shouldSend) {
      try {
        // 发送批次到Firehose
        await sendToFirehose(batch);
        // 发送成功后清空批次
        await env.LOG_BATCH_KV.put("log-batch", JSON.stringify({
          batch: [],
          createdAt: now.toISOString()
        }));
      } catch (err) {
        console.error("Firehose发送失败:", err);
        // 发送失败保留批次，下次请求重试
        await env.LOG_BATCH_KV.put("log-batch", JSON.stringify({ batch, createdAt }));
      }
    } else {
      // 未达发送条件，更新KV中的批次
      await env.LOG_BATCH_KV.put("log-batch", JSON.stringify({ batch, createdAt }));
    }

    return Response.redirect("http://newurl.com", 302);
  },
};

额外优化：分离日志发送与请求处理 #

如果担心发送日志的耗时影响重定向速度，可以搭配Cloudflare Cron Triggers（免费额度内可用），让定时任务负责批次发送，主Worker只做日志存储：

1. 创建一个定时触发的Worker，负责拉取KV批次并发送：

import { sendToFirehose } from "./lib/sendToFirehose";

export interface Env {
  AWS_ACCESS_KEY_ID: string;
  AWS_SECRET_ACCESS_KEY: string;
  LOG_BATCH_KV: KVNamespace;
}

export default {
  async scheduled(_event: ScheduledEvent, env: Env): Promise<void> {
    const batchRecord = await env.LOG_BATCH_KV.get("log-batch", "json") || { batch: [] };
    
    if (batchRecord.batch.length === 0) return;

    try {
      await sendToFirehose(batchRecord.batch);
      await env.LOG_BATCH_KV.put("log-batch", JSON.stringify({ batch: [] }));
    } catch (err) {
      console.error("定时任务发送批次失败:", err);
    }
  },
};

2. 主Worker只负责向KV添加日志，无需判断批次大小：

// 主Worker简化代码
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const logData = {
      date: new Date().toISOString(),
      url: request.url,  
    };

    let batchRecord = await env.LOG_BATCH_KV.get("log-batch", "json") || { batch: [] };
    batchRecord.batch.push(logData);
    await env.LOG_BATCH_KV.put("log-batch", JSON.stringify(batchRecord));

    return Response.redirect("http://newurl.com", 302);
  },
};

注意事项 #

• KV读写存在毫秒级延迟，但对于日志场景完全可接受；
• 并发请求下可能出现少量日志重复（多个实例同时读取到同一批次，都添加数据后存回），如果需要严格避免，可搭配Cloudflare Durable Objects做单例锁，但成本会略高；
• 确保sendToFirehose函数实现了正确的AWS签名，Cloudflare Worker环境下要注意时间同步问题（Worker环境时间是准确的，但需确保签名逻辑兼容）。

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