18.12-并发控制
要点
- Worker 是单线程异步模型——同 Isolate 内不会有真正的并行，但多个 Isolate 会并发访问共享资源
- 信号量（Semaphore）限制同时进行的任务数——防止下游系统被打爆
- 队列缓冲：任务过多时放进队列，按下游处理能力慢慢消费
- Worker 实例的并发限制：单 Isolate 同时处理的请求数受 CPU 时间限制
- Durable Objects 提供串行化——所有访问同一 DO 的请求排队执行
- Worker 没有 mutex，但 Durable Objects 天然串行化，KV/D1 有原子操作
1. Worker 的并发模型
Worker 是单线程异步模型——跟 Node.js 类似：
- 单线程：同一 Isolate 内，同一时刻只有一个 JavaScript 任务在执行
- 异步 I/O：网络请求、KV 读写等 I/O 操作异步进行，不阻塞主线程
- 多 Isolate：一个 Worker 在全世界多个数据中心有多个 Isolate 实例，彼此独立
这意味着：
- 同一 Isolate 内不需要 mutex——没有真正的并行
- 跨 Isolate 的并发控制需要借助 KV/D1/DO 等外部存储
- 下游系统（数据库、LLM API）面对的是多个 Isolate 的并发请求
2. 信号量（Semaphore）
2.1 场景
下游 API 有并发限制——比如 LLM API 每秒最多处理 10 个请求。超过会被拒绝（429）或拖慢响应。
用信号量限制同时进行的调用数：
// src/lib/semaphore.ts
export class Semaphore {
private running = 0
private queue: Array<() => void> = []
constructor(private maxConcurrent: number) {}
async acquire(): Promise<() => void> {
if (this.running < this.maxConcurrent) {
this.running++
return () => this.release()
}
// 满了，排队等待
return new Promise((resolve) => {
this.queue.push(() => {
this.running++
resolve(() => this.release())
})
})
}
private release(): void {
this.running--
const next = this.queue.shift()
if (next) next()
}
}2.2 使用
const llmSemaphore = new Semaphore(10) // 最多 10 个并发 LLM 调用
async function callLLMLimited(env: Env, params: LLMParams) {
const release = await llmSemaphore.acquire()
try {
return await env.AI.run(params.model, { messages: params.messages })
} finally {
release()
}
}注意：这个信号量是单 Isolate 的——10 个并发是指单个 Isolate 内。如果 Worker 有 100 个 Isolate，全局并发是 1000。
2.3 全局信号量
需要全局并发控制，用 Durable Objects 或 KV：
// 用 KV 做全局信号量（有误差）
async function acquireGlobalSemaphore(
kv: KVNamespace,
key: string,
max: number
): Promise<boolean> {
const current = parseInt((await kv.get(key)) || '0')
if (current >= max) return false
await kv.put(key, String(current + 1))
return true
}
async function releaseGlobalSemaphore(kv: KVNamespace, key: string) {
const current = parseInt((await kv.get(key)) || '0')
if (current > 0) {
await kv.put(key, String(current - 1))
}
}KV 不是原子的，高并发下会超出 max。严格限制需要 Durable Objects。
3. 队列缓冲
3.1 场景
任务突发来临（比如批量导入文档），但下游处理不过来。直接把所有请求都发给下游会把它打爆。
解决：任务先进队列，按固定速率消费。
3.2 Cloudflare Queue 实现
// 生产者：任务进队列
app.post('/api/embed/batch', async (c) => {
const { documents } = await c.req.json()
for (const doc of documents) {
await c.env.EMBED_QUEUE.send({
docId: doc.id,
text: doc.text,
})
}
return c.json({
message: `${documents.length} tasks queued`,
estimatedTime: Math.ceil(documents.length / 10), // 假设每秒处理 10 个
}, 202)
})
// 消费者：按固定速率消费
export async function embedConsumer(
batch: MessageBatch<EmbedTask>,
env: Env
) {
for (const msg of batch.messages) {
try {
await processEmbedTask(env, msg.body)
msg.ack()
} catch (e) {
msg.retry()
}
}
}Cloudflare Queue 的消费者有 max_batch_size 和 max_concurrency 配置——天然限流：
{
"queues": {
"consumers": [{
"queue": "embed-queue",
"max_batch_size": 10, // 每批最多 10 条
"max_concurrency": 5 // 最多 5 个并发批
}]
}
}4. Durable Objects 串行化
Durable Objects 的所有请求都排队执行——天然解决并发问题。
4.1 单实例串行
export class RateLimitedLLM {
private state: DurableObjectState
private queue: Array<{ request: Request; resolve: (r: Response) => void }> = []
private processing = false
constructor(state: DurableObjectState) {
this.state = state
}
async fetch(request: Request) {
return new Promise<Response>((resolve) => {
this.queue.push({ request, resolve })
this.processQueue()
})
}
private async processQueue() {
if (this.processing || this.queue.length === 0) return
this.processing = true
while (this.queue.length > 0) {
const { request, resolve } = this.queue.shift()!
try {
const result = await callLLM(request)
resolve(new Response(JSON.stringify(result)))
} catch (e) {
resolve(new Response(JSON.stringify({ error: 'LLM call failed' }), { status: 500 }))
}
}
this.processing = false
}
}所有对这个 DO 的请求按顺序执行——不会并发调用 LLM。
4.2 配合并发控制
DO 本身串行，但你可以控制并发度——比如允许 5 个并发：
export class ConcurrentLLM {
private state: DurableObjectState
private running = 0
private maxConcurrent = 5
private queue: Array<{ request: Request; resolve: (r: Response) => void }> = []
constructor(state: DurableObjectState) {
this.state = state
this.state.blockConcurrencyWhile(async () => {
const stored = await this.state.storage.get<number>('running')
if (stored !== undefined) this.running = stored
})
}
async fetch(request: Request) {
return new Promise<Response>((resolve) => {
this.queue.push({ request, resolve })
this.processQueue()
})
}
private async processQueue() {
while (this.running < this.maxConcurrent && this.queue.length > 0) {
const { request, resolve } = this.queue.shift()!
this.running++
await this.state.storage.put('running', this.running)
// 不 await，让它并发执行
;(async () => {
try {
const result = await callLLM(request)
resolve(new Response(JSON.stringify(result)))
} catch (e) {
resolve(new Response(JSON.stringify({ error: 'LLM call failed' }), { status: 500 }))
} finally {
this.running--
await this.state.storage.put('running', this.running)
this.processQueue()
}
})()
}
}
}5. Worker 实例并发限制
单个 Isolate 同时处理的请求数受 CPU 时间限制。如果每个请求都跑 CPU 密集型任务，Worker 会超时。
5.1 监控 Isolate 并发
// 用模块级变量记录当前请求数
let currentRequests = 0
export default {
async fetch(request: Request, env: Env) {
currentRequests++
try {
return await handleRequest(request, env)
} finally {
currentRequests--
}
},
}
// 暴露给监控
app.get('/admin/stats', (c) => {
return c.json({ currentRequests })
})5.2 自我保护
当前请求太多时，直接拒绝新请求：
export default {
async fetch(request: Request, env: Env) {
if (currentRequests > 50) {
return new Response('Too many requests', { status: 503 })
}
currentRequests++
try {
return await handleRequest(request, env)
} finally {
currentRequests--
}
},
}这个限制是单 Isolate 的。如果 Worker 有 100 个 Isolate，整体能处理 5000 个并发。
6. AbortController 与超时
长时间运行的任务要设超时，防止卡死：
async function callLLMWithTimeout(env: Env, params: LLMParams, timeoutMs: number) {
const controller = new AbortController()
const timeoutId = setTimeout(() => controller.abort(), timeoutMs)
try {
const result = await env.AI.run(params.model, {
messages: params.messages,
}, {
signal: controller.signal,
})
return result
} catch (e) {
if (e.name === 'AbortError') {
throw new Error(`LLM call timed out after ${timeoutMs}ms`)
}
throw e
} finally {
clearTimeout(timeoutId)
}
}7. 实战：批量 embedding 的并发控制
批量上传文档时，控制并发：
// src/lib/batch-embed.ts
export async function batchEmbed(
env: Env,
documents: { id: string; text: string }[],
concurrency = 5
): Promise<void> {
const semaphore = new Semaphore(concurrency)
await Promise.all(
documents.map(async (doc) => {
const release = await semaphore.acquire()
try {
const embedding = await embedWithCache(env, doc.text)
await env.VECTOR_INDEX.upsert([{
id: doc.id,
values: embedding,
}])
} finally {
release()
}
})
)
}5 个并发——避免同时发 1000 个 embedding 请求把 AI API 打爆。
8. 小结
Worker 环境的并发控制：
- 单 Isolate 单线程：不需要 mutex，但跨 Isolate 需要外部存储协调
- 信号量：限制同时进行的任务数——单 Isolate 用模块级 Semaphore，全局用 KV 或 Durable Objects
- 队列缓冲：任务进 Cloudflare Queue，按固定速率消费
- Durable Objects：天然串行化，可配置并发度
- Worker 实例保护：当前请求太多时返回 503
- 超时控制：用 AbortController 防止长时间任务卡死
批量任务场景常用信号量 + 队列组合——信号量控制并发度，队列吸收突发流量。
下一篇讲冷启动优化——Worker 冷启动延迟的来源、懒加载、预热、缓存关键依赖。