基于Cloudflare Workers的异步长时数据库查询最佳实践问询

问题背景 #

我希望通过Cloudflare Workers运行可能耗时数分钟的长时数据库查询，目前该工具在处理简单快速查询时表现出色，示例代码如下：

export default {
    async fetch(
        request: Request,
        env: Env,
        ctx: ExecutionContext
    ): Promise<Response> {
        const client = new Client(env.DB_URL);
        await client.connect();

        const result = await client.query("SELECT * FROM data");

        const resp = new Response(JSON.stringify(result.rows), {
            headers: { "Content-Type": "application/json" }
        });

        ctx.waitUntil(client.end());
        return resp;
    },
};

不想一直等待Worker完成长时查询，考虑采用Worker发起查询后将结果存入缓存，再通过轮询获取结果的方案，想知道是否有更符合最佳实践的解决方案。

推荐解决方案 #

1. 使用 Cloudflare Queues 实现异步任务处理 #

这是Cloudflare官方推荐的长时异步任务处理方案，适配查询耗时超过Worker请求超时限制（通常100秒左右）的场景：

• 生产者Worker：接收客户端请求，生成唯一任务ID，将查询参数和任务ID发送到Cloudflare Queue，立即返回任务ID给客户端。
• 消费者Worker：监听Queue中的任务，取出后执行长时数据库查询，完成后将结果存入Cloudflare KV或R2存储。
• 客户端：通过任务ID轮询KV/R2获取结果，或结合Webhook接收通知。

生产者代码示例：

export default {
    async fetch(request, env, ctx) {
        // 生成唯一任务ID
        const taskId = crypto.randomUUID();
        // 获取请求体中的查询语句
        const { query } = await request.json();
        
        // 将任务发送到指定队列
        await env.LONG_QUERY_QUEUE.send({
            taskId,
            query
        });

        return new Response(JSON.stringify({ taskId }), {
            headers: { "Content-Type": "application/json" }
        });
    }
};

消费者代码示例：

export default {
    async queue(batch, env) {
        for (const message of batch.messages) {
            const { taskId, query } = message.body;
            const client = new Client(env.DB_URL);
            
            try {
                await client.connect();
                const result = await client.query(query);
                // 将查询结果存入KV存储
                await env.QUERY_RESULTS.put(taskId, JSON.stringify(result.rows));
            } catch (err) {
                // 捕获错误并存储错误信息
                await env.QUERY_RESULTS.put(`${taskId}_error`, err.message);
            } finally {
                await client.end();
            }

            // 标记消息已处理完成
            message.ack();
        }
    }
};

2. 优化轮询方案：结合 KV + waitUntil #

如果查询耗时未超过Worker超时限制，但仍希望避免客户端长时间等待，可以优化你原本的思路：

• 生成唯一任务ID，立即返回给客户端
• 使用ctx.waitUntil()在后台异步执行查询，完成后将结果写入KV
• 客户端通过任务ID轮询KV获取结果

代码示例：

export default {
    async fetch(request, env, ctx) {
        const taskId = crypto.randomUUID();
        // 从URL参数中获取查询语句
        const url = new URL(request.url);
        const query = url.searchParams.get("query");

        // 后台执行查询并存储结果
        ctx.waitUntil((async () => {
            const client = new Client(env.DB_URL);
            try {
                await client.connect();
                const result = await client.query(query);
                await env.QUERY_RESULTS.put(taskId, JSON.stringify(result.rows));
            } catch (err) {
                await env.QUERY_RESULTS.put(`${taskId}_error`, err.message);
            } finally {
                await client.end();
            }
        })());

        // 立即返回任务ID，无需等待查询完成
        return new Response(JSON.stringify({ taskId }), {
            headers: { "Content-Type": "application/json" }
        });
    }
};

3. 使用 Webhook 替代轮询（减少无效请求） #

如果客户端可以提供回调URL，完全可以避免轮询带来的无效请求：

• 客户端请求时携带webhook_url参数
• Worker将任务放入Queue后立即返回任务ID
• 消费者Worker完成查询后，主动POST结果到指定的webhook_url

消费者代码片段：

// 在消费者Worker的queue处理函数中
const { taskId, query, webhookUrl } = message.body;
// ...执行查询逻辑后
await fetch(webhookUrl, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
        taskId,
        result: result.rows,
        status: "success"
    })
});

方案选择建议 #

• 查询耗时超过100秒：优先用Cloudflare Queues，规避Worker请求超时限制。
• 查询耗时在超时内但不想让客户端等待：用KV + waitUntil的优化方案，实现成本低。
• 客户端支持回调：用Webhook替代轮询，减少不必要的请求开销。

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