<!-- title: 并发批处理Worker池实战 -->

# Claude Code 并发批处理 Worker 池实战：用 Hooks 护栏驱动 `-p` 批量扇出

## 概述

本教程聚焦 Claude Code 的自动化三件套——无头 `-p` 模式、Hooks 事件护栏、批量并发执行——并将它们组装成一个**可投产的 Worker 池**：一条任务队列被分发到 N 个并发 `claude -p` 进程，每个进程由同一套 Hooks 策略约束（禁止越权写、自动格式化、回合预算熔断），最后由 `stream-json` 输出汇聚成一份汇总报告。适合需要在 CI、夜间批跑、多仓库批量改造等场景下"一次跑几十个独立任务"的工程师。

## 前置条件

- 已安装 Claude Code（`claude --version` 可输出 ≥ 1.x）
- 具备 Bash / Shell 脚本基础，了解管道、`xargs -P` 或 GNU `parallel`
- 一台 Linux/macOS 机器，能访问 Anthropic API 或已 `claude login`
- 了解 Git 基础（本教程用 `git worktree` 做并发隔离）
- 建议预先阅读"无头模式结构化输出"与"钩子事件总线"两篇，本文在其之上做编排层封装

---

## 一、核心心智模型：确定性护栏 + 并发扇出

很多团队第一次用 `claude -p` 跑批量任务时，会踩两类坑：

1. **并发写冲突**：多个 `claude -p` 进程改同一个工作区，互相覆盖。
2. **护栏缺失**：无人值守时模型越权执行 `rm -rf`、`git push --force`、写密钥。

解法是把职责拆成两层：

| 层 | 载体 | 职责 |
|---|---|---|
| **执行层** | `claude -p` 进程池 | 承载一个独立任务，产生 `stream-json` 结果 |
| **策略层** | `.claude/settings.json` 的 Hooks | 在每个 Worker 内部拦截/放行/熔断 |
| **调度层** | Shell 脚本（`xargs -P` / `parallel`） | 拆队列、控并发、隔离工作区、汇聚结果 |

关键洞察：**Hooks 是"声明一次、对每个 Worker 进程都生效"的策略**。你不需要在每个 prompt 里反复叮嘱"别 force push"，而是把它写进 `PreToolUse` 护栏，由 harness 在每次工具调用前强制执行。这就是"确定性"——它不依赖模型的"自觉"。

```mermaid
flowchart LR
    Q[任务队列 jobs.jsonl] --> SCHED[调度器 xargs -P N]
    SCHED --> W1[Worker 1 claude -p]
    SCHED --> W2[Worker 2 claude -p]
    SCHED --> WN[Worker N claude -p]
    W1 -.策略.-> H[(Hooks settings.json)]
    W2 -.策略.-> H
    WN -.策略.-> H
    H -->|PreToolUse 拦截| GUARD{放行/阻断/提醒}
    W1 --> R1[stream-json 结果]
    W2 --> R2[stream-json 结果]
    WN --> RN[stream-json 结果]
    R1 --> AGG[结果汇聚 report.md]
    R2 --> AGG
    RN --> AGG
```

## 二、`-p` 无头模式：单 Worker 的解剖

### 2.1 三种输出格式

```bash
# text：纯文本，最简单
claude -p "把 README 里所有的 TODO 收集成一个列表" 

# json：单条 JSON 对象，便于 jq 解析
claude -p "..." --output-format json | jq -r '.result'

# stream-json：换行分隔的事件流，批处理的黄金格式
claude -p "..." --output-format stream-json
```

`stream-json` 会按时间顺序吐出几类事件，批处理最关心的是首尾两个：

- **`system` 事件**（subtype `init`）：包含 `session_id`，**这是断点续跑的钥匙**。
- **`result` 事件**（subtype `success`/`error`）：包含 `result`（最终文本）、`cost_usd`、`duration_ms`、`is_error`、`num_turns`。

一个最小的事件抽取器：

```bash
# 从 stream-json 抓出 session_id 和最终结果
claude -p "修复 utils.py 里的类型错误" \
  --output-format stream-json \
  --allowedTools "Edit,Read,Bash(pytest *)" \
  > /tmp/w1.jsonl

# 取 session_id（用于 --resume）
grep '"type":"system"' /tmp/w1.jsonl | head -1 | jq -r '.session_id'

# 取最终结果与成本
grep '"type":"result"' /tmp/w1.jsonl | jq -r '"cost=\(.cost_usd) turns=\(.num_turns)\n\(.result)"'
```

### 2.2 工具白名单是第一道护栏

`--allowedTools` 限定 Worker 只能用列出的工具，逗号分隔，支持通配：

```bash
claude -p "..." \
  --allowedTools "Read,Edit,Write,Bash(git *),Bash(npm test)"
```

> **常见问题**：Worker 报"Tool not allowed"。原因：prompt 触发了一个未在白名单里的工具。解决：要么放宽白名单，要么改 prompt 让模型不依赖该工具。生产中建议**先收紧、再按需放开**。

### 2.3 `--max-turns` 给单 Worker 设硬上限

无人值守必须设回合预算，否则一个 Worker 卡在死循环能烧掉一整夜的额度：

```bash
claude -p "重构这个模块" --max-turns 25 --output-format stream-json
```

`--max-turns 25` 会在第 25 个回合强制终止，`result` 事件 subtype 会标记非成功。

### 2.4 断点续跑：`--resume` 与 `--continue`

```bash
# 第一次跑，记住 session_id
SID=$(claude -p "..." --output-format stream-json | \
      grep '"type":"system"' | jq -r '.session_id' | head -1)

# 之后基于这个会话继续（保留全部上下文）
claude -p "继续刚才的重构，跑一下测试" --resume "$SID" --output-format stream-json
```

`--continue`（`-c`）则直接续最近一次会话，适合交互式补刀；批处理里更推荐显式 `--resume <id>`，避免多 Worker 抢同一个"最近会话"。

---

## 三、Hooks 策略层：每个 Worker 的内部刹车

Hooks 定义在 `.claude/settings.json`（项目级）或 `~/.claude/settings.json`（用户级）。批处理时，我们通常把策略放在**队列共享的项目根 `.claude/settings.json`**，让每个 `claude -p` 进程自动继承。

### 3.1 配置骨架

```json
{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [{ "type": "command", "command": ".claude/hooks/guard-bash.sh" }]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "Edit|Write",
        "hooks": [{ "type": "command", "command": ".claude/hooks/auto-fmt.sh" }]
      }
    ],
    "Stop": [
      {
        "hooks": [{ "type": "command", "command": ".claude/hooks/budget-guard.sh" }]
      }
    ]
  }
}
```

`matcher` 是**正则**，匹配工具名（`Bash`、`Edit`、`Write`、`Read` 等），留空表示匹配所有工具。

### 3.2 Hook 的输入与退出码语义

每个 hook 进程通过 **stdin 收到一段 JSON**，关键字段：

```json
{
  "session_id": "abc123",
  "transcript_path": "/path/to/transcript.jsonl",
  "cwd": "/repo/worktree-3",
  "hook_event_name": "PreToolUse",
  "tool_name": "Bash",
  "tool_input": { "command": "git push --force origin main", "description": "..." }
}
```

退出码决定行为：

| 退出码 | 含义（PreToolUse） |
|---|---|
| `0` | 放行；stdout 可作为额外上下文喂给模型 |
| `2` | **阻断**该工具调用，stderr 内容会回显给模型（让它知道为什么被拦） |
| 其他非零 | 非阻断的错误，harness 会告警但继续 |

### 3.3 实战护栏 1：禁止危险 Bash（`PreToolUse`）

`.claude/hooks/guard-bash.sh`：

```bash
#!/usr/bin/env bash
# 读取 hook 输入
input=$(cat)
cmd=$(echo "$input" | jq -r '.tool_input.command // ""')

# 危险模式黑名单（确定性拦截，不依赖模型判断）
deny_patterns='rm[[:space:]]+-rf|git[[:space:]]+push.*--force|git[[:space:]]+push.*-f|>[[:space:]]*/dev/sd|chmod[[:space:]]+-R[[:space:]]+777'

if echo "$cmd" | grep -Eq "$deny_patterns"; then
  echo "⛔ 被护栏拦截：检测到危险命令模式 → $cmd" >&2
  exit 2   # 阻断，stderr 回显给模型
fi

# 放行
exit 0
```

`chmod +x` 之后，任何 Worker 试图跑 `git push --force` 都会被挡下，模型会收到"被护栏拦截"的反馈并自行改路。

> **常见坑**：很多人在 `Edit`/`Write` 上也用 `exit 2` 想阻止写入，但忘了 `Edit` 的 `tool_input` 结构不同（有 `file_path`/`old_string`/`new_string`）。匹配器要用对应的字段，否则永远拦不住。

### 3.4 实战护栏 2：写后自动格式化（`PostToolUse`）

`.claude/hooks/auto-fmt.sh`：

```bash
#!/usr/bin/env bash
input=$(cat)
file=$(echo "$input" | jq -r '.tool_input.file_path // ""')

[ -z "$file" ] || [ ! -f "$file" ] && exit 0

case "$file" in
  *.py)   ruff format "$file" 2>/dev/null; ruff check --fix "$file" 2>/dev/null ;;
  *.ts|*.tsx|*.js) npx --no-install prettier --write "$file" 2>/dev/null ;;
  *.go)   gofmt -w "$file" 2>/dev/null ;;
esac
exit 0   # PostToolUse 用 0，格式化失败也不阻断
```

这样每个 Worker 改完文件立即格式化，批跑结束后仓库就是干净的，无需单独跑 lint。

### 3.5 实战护栏 3：回合预算熔断（`Stop`）

`Stop` 事件在模型准备结束回合时触发。我们可以在这里读取累计成本，超阈值就**强制继续**（`continue: false` 配合 reason），或者干脆记录告警。

`.claude/hooks/budget-guard.sh`（轻量版：只告警不强行续）：

```bash
#!/usr/bin/env bash
input=$(cat)
transcript=$(echo "$input" | jq -r '.transcript_path')

# 从 transcript 估算 token 用量，超阈值就输出一个醒目提醒
# stdout 会被注入为额外上下文（exit 0 时）
echo "⏱️ Worker 即将停止，请确认任务已完成且测试通过。" 
exit 0
```

若要**硬性强制模型继续工作**直到测试通过，可以让脚本输出 JSON：

```bash
# 强制继续（让模型别急着停）
printf '{"continue": true, "stopReason": "测试尚未全部通过，继续工作"}'
exit 0
```

> **注意**：`continue: true` 配合"必须满足某条件才停"的 prompt 容易造成无限循环。务必同时用 `--max-turns` 做硬上限兜底。

### 3.6 高级：用 stdout JSON 精确控制

除了退出码，hook 还能向 stdout 输出结构化 JSON 做更精细控制，关键字段：

```json
{
  "continue": false,
  "stopReason": "预算耗尽",
  "suppressOutput": false,
  "additionalContext": "（注入到对话的额外上下文）"
}
```

例如 `UserPromptSubmit` hook 可以给每个 Worker 的输入注入共享上下文（编码规范、文件清单），避免在每个 prompt 里重复粘贴。

---

## 四、调度层：把任务队列扇出到 Worker 池

### 4.1 任务队列格式

`jobs.jsonl`，每行一个独立任务：

```json
{"id":"auth","repo":"https://github.com/acme/svc-auth.git","task":"把所有 print 换成 logging，并补类型注解"}
{"id":"billing","repo":"https://github.com/acme/svc-billing.git","task":"为 /invoice 接口补 pytest 覆盖到 80%"}
{"id":"gateway","repo":"https://github.com/acme/svc-gateway.git","task":"把 requests 替换为 httpx，保持行为不变"}
```

每行一个任务的好处：`jq`/`xargs` 处理天然友好，断点续跑只需删掉已完成的行。

### 4.2 单 Worker 封装脚本

`worker.sh` 接收一行 job，做四件事：克隆/切 worktree → 跑 `claude -p`（带护栏与预算）→ 抽取结果 → 写报告。

```bash
#!/usr/bin/env bash
# worker.sh —— 单个 Worker 的生命周期
set -u
LINE="$1"
OUT_DIR="${2:-./out}"

id=$(echo "$LINE"   | jq -r '.id')
repo=$(echo "$LINE"  | jq -r '.repo')
task=$(echo "$LINE"  | jq -r '.task')

WORK=".work/$id"
LOG="$OUT_DIR/$id.jsonl"
mkdir -p "$WORK" "$OUT_DIR"

# 1) 获取代码（已存在则 fetch）
if [ -d "$WORK/.git" ]; then
  git -C "$WORK" fetch --quiet && git -C "$WORK" reset --hard origin/HEAD -q
else
  git clone --quiet --depth 1 "$repo" "$WORK"
fi

# 把共享 hooks 策略软链进去（策略唯一来源）
mkdir -p "$WORK/.claude/hooks"
ln -sf "$(pwd)/.claude/settings.json" "$WORK/.claude/settings.json"
ln -sf "$(pwd)/.claude/hooks/"* "$WORK/.claude/hooks/" 2>/dev/null || true

# 2) 跑 Worker（带工具白名单 + 回合预算 + stream-json）
cd "$WORK"
claude -p "$task" \
  --output-format stream-json \
  --allowedTools "Read,Edit,Write,Bash(git *),Bash(pytest*),Bash(npm test)" \
  --max-turns 30 \
  > "$OUT_DIR/$id.jsonl" 2> "$OUT_DIR/$id.err"

cd - >/dev/null

# 3) 抽取结果摘要
sid=$(grep '"type":"system"' "$LOG" | head -1 | jq -r '.session_id')
result=$(grep '"type":"result"' "$LOG" | head -1)
cost=$(echo "$result" | jq -r '.cost_usd // "?"')
turns=$(echo "$result" | jq -r '.num_turns // "?"')
ok=$(echo "$result" | jq -r '.is_error // "true"')

# 4) 汇报行（供后续聚合）
printf '%s\t%s\t%s\t%s\t%s\n' "$id" "$ok" "$cost" "$turns" "$sid" >> "$OUT_DIR/_summary.tsv"
echo "[$id] done ok=$ok cost=\$$cost turns=$turns session=$sid"
```

> **隔离要点**：每个 Worker 用独立的 `.work/<id>` 目录克隆（或 `git worktree add`），从根上避免多进程写同一棵树。`git clone --depth 1` 让大批仓库快速就绪。

### 4.3 并发扇出：`xargs -P` 与 `parallel` 两种姿势

```bash
# 姿势 A：xargs -P（POSIX 通用）
mkdir -p out
cat jobs.jsonl | xargs -P 4 -I{} ./worker.sh {} ./out

# 姿势 B：GNU parallel（输出更整齐，自带进度）
mkdir -p out
parallel --bar -j 4 ./worker.sh {} ./out :::: jobs.jsonl
```

`-P 4` / `-j 4` = 最多 4 个 Worker 同时跑。并发度选择经验：

- **看 IO**：每个 Worker 是独立 `claude` 进程，主要等 API；CPU 占用低，并发瓶颈在网络与速率限制。
- **看速率配额**：并发太高会触发 API 速率限制，建议从 4–8 起步，观察 `429` 后回调。
- **看成本**：4 并发 × 30 回合的单夜批跑，很容易上百美元。先小队列试跑，`_summary.tsv` 看单任务成本再放量。

### 4.4 汇聚报告

```bash
# 把 _summary.tsv 渲染成 Markdown 总表
{
  echo "# 批处理报告"
  echo
  echo "| 任务 | 成功 | 成本(\$) | 回合数 | session_id |"
  echo "|---|---|---|---|---|"
  sort out/_summary.tsv | awk -F'\t' '{printf "| %s | %s | %s | %s | `%s` |\n",$1,$2,$3,$4,$5}'
  echo
  total=$(awk -F'\t' '{s+=$3} END{print s}' out/_summary.tsv)
  echo "**总成本：\$${total}**"
} > report.md
```

## 五、完整实战：跨 3 个仓库的并发代码改造

把上述拼成一个一键脚本 `batch.sh`：

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

JOBS="${1:-jobs.jsonl}"
PARALLEL="${2:-4}"
rm -rf out .work && mkdir -p out .work

echo "→ 扇出 $PARALLEL 个 Worker 处理 $(wc -l < "$JOBS") 个任务"
parallel -j "$PARALLEL" ./worker.sh {} ./out :::: "$JOBS"

echo "→ 汇聚报告"
./report.sh        # 即上一节的 awk 片段
echo "完成，见 report.md"
```

**预期结果**：`out/<id>.jsonl` 各自记录完整事件流；`out/_summary.tsv` 汇总每任务成败/成本/会话；`report.md` 是给人看的总表。任一任务失败（`is_error: true`），可直接用其 `session_id` 续跑：

```bash
claude -p "上一次中断了，继续完成并把测试跑绿" --resume <session_id>
```

> **常见问题**：部分 Worker 报 `is_error`。先看 `out/<id>.err` 与 transcript 末尾。九成是护栏阻断（说明策略生效了）或 `--allowedTools` 太紧（放宽）、或单任务超 `--max-turns`（拆任务或上调预算）。

## 六、安全与成本红线

| 维度 | 手段 |
|---|---|
| 越权写 | `PreToolUse` 黑名单 `rm -rf` / `git push -f` / 写 `/etc` |
| 密钥泄露 | `PreToolUse` 对 `Write`/`Bash` 扫描 `AKIA*`、`sk-`、`-----BEGIN` 等模式，命中即 `exit 2` |
| 预算失控 | 每个 Worker `--max-turns` + `Stop` hook 双保险 |
| 并发写冲突 | 每 Worker 独立 worktree/clone，绝不共享工作区 |
| 速率限制 | 并发从 4 起步，监控 `429`，`xargs`/`parallel` 失败自动重排 |
| 可审计 | 全量保留 `*.jsonl` 事件流，含每步工具调用与成本 |

## 七、进阶：用 Subagent 在单 Worker 内并行

当**单个任务**内部有可并行的子项（如一个大仓里改 10 个互不相关的模块），不必起 10 个 `claude -p`，而是在一个 Worker 内用 Subagent 扇出。在仓库的 `.claude/agents/` 下定义：

```yaml
# .claude/agents/refactor-worker.yaml
---
name: refactor-worker
description: 独立重构一个模块并跑测试，用于批量子任务并行
tools: Read, Edit, Write, Bash(git *), Bash(pytest*)
model: sonnet
---
你是模块重构专员。只改分配给你的模块目录，完成后必须运行测试并报告结果。
不要触碰其他模块。
```

主 Worker（`claude -p`）在 prompt 里被指示"对每个模块派发一个 refactor-worker 子代理并行处理"，harness 会并发调度它们。**子代理与外层 Worker 池的区别**：子代理共享同一仓库上下文但有独立 token 窗口，适合"同仓多模块"；外层 Worker 池适合"多仓多任务"。两者可叠加：外层多仓 × 内层多模块。

## 八、何时该用哪一层

```mermaid
flowchart TD
    A[批量任务需求] --> B{任务间共享代码库?}
    B -- 多个独立仓库 --> C[外层 Worker 池<br/>xargs -P + claude -p]
    B -- 同一仓库多模块 --> D[Subagent 内层扇出]
    A --> E{需要确定性安全约束?}
    E -- 是 --> F[PreToolUse/PostToolUse Hooks]
    E -- 仅需预算控制 --> G[Stop hook + --max-turns]
    C --> F
    D --> F
    F --> Z[汇聚 stream-json → report]
    G --> Z
```

## 延伸阅读

- **Claude Code Headless 官方文档**：`-p` 全部 flag 与三种 `--output-format` 的权威说明，含 `--json-schema` 结构化输出
- **Hooks Reference 官方文档**：全部生命周期事件、stdin JSON schema、退出码与 stdout 控制字段
- **Subagents 官方文档**：`.claude/agents/*.yaml` frontmatter 字段与并行调度语义
- **Git Worktree**：`git worktree add` 在同一仓库创建多个独立工作目录，是并发隔离的轻量替代
- **GNU Parallel 教程**：`-j`、`--bar`、`--resume --joblog` 实现"失败重跑"的批处理调度

## 训练营适用性评估

- **适用类型**：DevOps/SRE 训练营、AI 工程化训练营、研发效能训练营。也适合"AI 辅助大规模代码迁移""技术债批量偿还"专题工作坊。
- **前置基础**：学员需先掌握 `claude -p` 单进程用法与 settings.json 基本配置（建议先修"无头模式结构化输出"与"钩子事件总线"两节）。
- **教学时长**：建议 4–5 学时。1 学时讲心智模型与 `-p` 解剖；1.5 学时动手写三个 Hook 护栏；1.5 学时搭建 Worker 池并跑通 3 仓库小队列；0.5–1 学时讨论成本控制与故障续跑。
- **学员产出**：一个可带回的 `batch.sh` + 护栏集合，能直接接入自己团队的批量改造任务。

## 参考来源

1. [Run Claude Code programmatically（Headless 官方文档）](https://code.claude.com/docs/en/headless) — `-p`、`--output-format stream-json`、`--resume`、`--allowedTools` 的权威定义
2. [Hooks reference（Claude Code 官方文档）](https://code.claude.com/docs/en/hooks) — 生命周期事件、matcher 正则、stdin JSON、退出码 2 阻断语义
3. [Create custom subagents（Claude Code 官方文档）](https://code.claude.com/docs/en/sub-agents) — `.claude/agents/*.yaml` frontmatter 与并行子代理调度
4. [Claude Code Headless Mode: The Complete Self-Hosting Guide (2026)](https://amux.io/guides/claude-code-headless/) — CI/CD 与夜间批跑的 auth、成本追踪、并发模式
5. [How Do I Run Parallel Batch Operations with Claude Code CLI?](https://docs.bswen.com/blog/2026-03-21-claude-code-parallel-batch-operations/) — `xargs -P` / `parallel` 配合 `--allowedTools` 的并发实战与速率控制经验

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)