工作窃取

概念定义

单一共享任务队列会让每次提交与取出都在同一把锁上竞争。工作窃取去中心化：每个 worker 拥有本地双端队列，正常在本端压入弹出，只有队列空了才从别人队列的远端偷取。

工作机制

本地端只有属主访问——无争用、成本极低；偷取端只在“走投无路者”之间被触碰——负载尚可时本就稀有。受害者选择按随机或按规模扫描，避免雷群式偷取。任务必须可拆分，偷来的才有价值。

使用场景

子问题不均的并行计算：递归分治、混合体量的 JSON 解析、编译器流水线，以及平衡 continuation 的异步运行时。ForkJoinPool 直接暴露它，parallel stream 的 commonPool 隐式乘坐它。

常见误区

窃取带来的是统计意义上的均衡而非保证，偏斜的任务体量仍会留缝。它也没有消灭同步——锁还在队列两端，只是把争用压到稀有路径。“只利 CPU 密集”是误解：异步任务的 continuation、并行 GC 的工作都在受益。任务过小偷取流量主导，任务过大偷了没用；worker 里写阻塞操作会让队列无法让渡——异步运行时才把窃取与协作式让点配对。

代码示例

#!/usr/bin/env bash
set -euo pipefail

lock_file=${1:-"${TMPDIR:-/tmp}/devexamples-backup.lock"}
max_wait=${LOCK_WAIT_SECONDS:-5}
work_file=""

cleanup() {
    local status=$?
    if [[ -n "$work_file" && -f "$work_file" ]]; then
        rm -f -- "$work_file"
    fi
    flock -u 9 2>/dev/null || true
    exit "$status"
}
trap cleanup EXIT

exec 9>"$lock_file"

if ! flock -w "$max_wait" 9; then
    printf 'could not acquire %s within %ss\n' "$lock_file" "$max_wait" >&2
    exit 75
fi

printf 'lock acquired by pid %d\n' "$"

work_file="$(mktemp "${TMPDIR:-/tmp}/backup.XXXXXX")"
printf 'job %d started at %s\n' "$" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" > "$work_file"

sleep "${JOB_SECONDS:-1}"
cat -- "$work_file"
printf 'job %d finished; lock released on exit\n' "$"

import asyncio
from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    identifier: int
    payload: str


async def worker(
    name: str,
    queue: "asyncio.Queue[Job | None]",
    results: list[tuple[str, int]],
) -> None:
    while True:
        job = await queue.get()
        try:
            if job is None:
                return
            await asyncio.sleep(0.01)
            results.append((name, job.identifier))
        finally:
            queue.task_done()


async def run_jobs(jobs: list[Job], workers: int = 3) -> list[tuple[str, int]]:
    if workers < 1:
        raise ValueError("workers must be positive")
    queue: asyncio.Queue[Job | None] = asyncio.Queue(maxsize=workers * 2)
    results: list[tuple[str, int]] = []
    tasks = [
        asyncio.create_task(worker(f"worker-{index}", queue, results))
        for index in range(workers)
    ]
    for job in jobs:
        await queue.put(job)
    for _ in tasks:
        await queue.put(None)
    await queue.join()
    await asyncio.gather(*tasks)
    return results


def main() -> None:
    jobs = [Job(index, f"payload-{index}") for index in range(8)]
    results = asyncio.run(run_jobs(jobs))
    print(f"processed {len(results)} jobs")
    for name, identifier in sorted(results, key=lambda row: row[1]):
        print(name, "->", identifier)
    assert sorted(identifier for _, identifier in results) == list(range(8))


if __name__ == "__main__":
    main()