多进程并发控制 —— 进程数限制、CPU/内存监控、优雅降级
简介
在前面的文章中，我们学习了 Codex CLI 的工作树并行（）—— 通过
git worktree创建多个独立工作目录，每个目录运行一个独立的 Codex 实例，实现真正的并行开发。
但并行开发带来了一个不可避免的问题：
资源是有限的，你不能无限地启动 Codex 实例。
如果你在一台 8 核 16GB 的服务器上同时启动 20 个 Codex 实例，结果会怎样？
- CPU 使用率飙到 100%，每个进程都抢不到时间片
- 内存耗尽，系统开始使用 swap，性能暴跌
- 部分进程被 OOM Killer 杀死，任务失败且没有通知
- 网络请求排队，API 调用超时
并发控制的本质，是在"吞吐量"和"稳定性"之间找到平衡点。
本文将系统地介绍如何在多 Codex 实例场景下实施并发控制：进程数限制、CPU/内存监控、优雅降级策略，以及完整的自动化管理脚本。
一、为什么需要并发控制
1.1 无限制的并发 = 灾难
想象一个场景：你有 50 个 PR 需要 AI Review。
# ❌ 错误做法：一次性启动 50 个 Codex 实例
for pr in $(gh pr list --json number -q '.[].number'); do
codex exec --full-auto "Review PR #$pr" &
done
wait这个脚本会在几秒钟内创建 50 个后台进程。每个 Codex 实例：
- 需要加载项目上下文（消耗 CPU）
- 发送 API 请求（占用网络带宽）
- 处理大文件内容（消耗内存）
- 可能执行测试（额外的 CPU/内存开销）
┌───────────────────────────────────────────────────┐
│ 50 个 Codex 实例同时运行的结果 │
├───────────────────┬───────────────────────────────┤
│ 指标 │ 结果 │
├───────────────────┼───────────────────────────────┤
│ CPU 使用率 │ 100%（系统级争抢） │
│ 内存使用 │ 超出物理内存，触发 OOM │
│ API 请求排队 │ 超过速率限制，大量 429 错误 │
│ 网络 I/O │ 带宽占满，SSH 都卡顿 │
│ 磁盘 I/O │ 大量临时文件写入，IOPS 打满 │
│ 最终结果 │ 45/50 失败，5/50 勉强完成 │
└───────────────────┴───────────────────────────────┘1.2 并发控制的三个维度
┌──────────────────────────────────────────────┐
│ 并发控制三维度 │
├──────────────────────────────────────────────┤
│ │
│ 1. 进程数限制（Concurrency Limit） │
│ ├── 固定上限 │
│ ├── 动态调整 │
│ └── 队列管理 │
│ │
│ 2. 系统资源监控（Resource Monitoring） │
│ ├── CPU 使用率 │
│ ├── 内存使用率 │
│ ├── 磁盘 I/O │
│ └── 网络带宽 │
│ │
│ 3. 优雅降级（Graceful Degradation） │
│ ├── 自动暂停 │
│ ├── 任务优先级调整 │
│ └── 失败重试 │
│ │
└──────────────────────────────────────────────┘二、进程数限制
2.1 基于固定上限的并发控制
最简单也是最有效的方法：设置一个固定的最大并发数。
#!/bin/bash
# concurrent-codex-fixed.sh —— 固定并发数控制
set -euo pipefail
# 配置参数
MAX_CONCURRENT=${CODEX_MAX_CONCURRENT:-4} # 最大并发数
JOB_QUEUE=() # 任务队列
RUNNING_PIDS=() # 运行中的进程 PID
# 日志函数
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}
# 获取当前运行的 Codex 进程数
get_running_count() {
local count=0
for pid in "${RUNNING_PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
((count++))
fi
done
echo "$count"
}
# 清理已完成的进程 PID
cleanup_pids() {
local new_pids=()
for pid in "${RUNNING_PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
new_pids+=("$pid")
fi
done
RUNNING_PIDS=("${new_pids[@]+"${new_pids[@]}"}")
}
# 等待有空闲槽位
wait_for_slot() {
while true; do
cleanup_pids
local running=$(get_running_count)
if [ "$running" -lt "$MAX_CONCURRENT" ]; then
return 0
fi
sleep 2
done
}
# 执行单个任务
run_task() {
local task_id=$1
local worktree_path=$2
local task_desc=$3
log "Starting task $task_id: $task_desc"
log " Worktree: $worktree_path"
cd "$worktree_path"
# 运行 Codex，记录输出
codex exec --full-auto "$task_desc" \
--output-format json \
> "/tmp/codex-task-${task_id}.json" 2>&1
local exit_code=$?
log "Task $task_id completed with exit code: $exit_code"
return $exit_code
}
# 主循环：处理任务队列
process_queue() {
local total=${#JOB_QUEUE[@]}
local completed=0
for task in "${JOB_QUEUE[@]}"; do
wait_for_slot
# 解析任务
IFS='|' read -r task_id worktree_path task_desc <<< "$task"
# 启动后台进程
run_task "$task_id" "$worktree_path" "$task_desc" &
RUNNING_PIDS+=($!)
log " → PID ${RUNNING_PIDS[-1]} | Running: $(get_running_count)/$MAX_CONCURRENT"
((completed++))
done
# 等待所有任务完成
wait
log "All $total tasks completed"
}
# 示例：批量 PR Review
setup_pr_review_tasks() {
local repo_path="/opt/data/my-project"
local wt_base="/tmp/codex-wt"
mkdir -p "$wt_base"
local task_id=0
for pr_number in $(gh pr list --state open --json number -q '.[].number'); do
# 为每个 PR 创建独立的 worktree
local wt_path="${wt_base}/pr-${pr_number}"
if [ ! -d "$wt_path" ]; then
(cd "$repo_path" && git worktree add "$wt_path" main)
fi
JOB_QUEUE+=("${task_id}|${wt_path}|Review PR #${pr_number}: check for bugs, security issues, and style violations")
((task_id++))
done
}
# 入口
setup_pr_review_tasks
process_queue2.2 基于 CPU 核心的动态并发数
更好的策略是根据服务器的 CPU 核心数自动设置并发上限：
#!/bin/bash
# concurrent-codex-dynamic.sh —— 基于 CPU 核心数的动态并发
set -euo pipefail
# 获取逻辑 CPU 核心数
CPU_CORES=$(nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || echo 4)
# 推荐并发策略
# - 每个 Codex 实例大约消耗 1 个 CPU 核心的算力
# - 保留 1-2 个核心给系统和监控
RECOMMENDED_CONCURRENT=$(( CPU_CORES - 2 ))
if [ "$RECOMMENDED_CONCURRENT" -lt 1 ]; then
RECOMMENDED_CONCURRENT=1
fi
if [ "$RECOMMENDED_CONCURRENT" -gt 8 ]; then
RECOMMENDED_CONCURRENT=8 # 上限 8，避免 API 速率限制
fi
log "CPU cores: $CPU_CORES"
log "Recommended max concurrent: $RECOMMENDED_CONCURRENT"
log "Actual max concurrent: $MAX_CONCURRENT"2.3 使用 GNU parallel 简化并发
如果你有 GNU parallel 工具，可以大幅简化并发控制：
#!/bin/bash
# parallel-codex.sh —— 使用 GNU parallel
set -euo pipefail
# 创建任务列表文件
TASK_FILE="/tmp/codex-tasks.txt"
> "$TASK_FILE"
# 生成任务
for pr_number in $(gh pr list --state open --json number -q '.[].number'); do
echo "Review PR #${pr_number}: check for bugs, security issues, and style violations" >> "$TASK_FILE"
done
# 使用 parallel 控制并发
# --jobs: 最大并发数
# --joblog: 记录执行日志
# --eta: 显示预计完成时间
# --halt: 失败时停止
parallel \
--jobs 4 \
--joblog /tmp/codex-parallel.log \
--eta \
'cd /opt/data/my-project && codex exec --full-auto "{}"' \
:::: "$TASK_FILE"
echo "All tasks complete. Check log: /tmp/codex-parallel.log"2.4 使用 tmux 编排并发会话
结合前文介绍的后台模式，可以用 tmux 管理多个 Codex 会话：
#!/bin/bash
# tmux-concurrent-codex.sh —— tmux 编排多 Codex 并发会话
set -euo pipefail
SESSION_NAME="codex-batch"
MAX_WINDOWS=4
# 创建 tmux 会话
tmux new-session -d -s "$SESSION_NAME"
# 拆分为多个窗口
for i in $(seq 1 $((MAX_WINDOWS - 1))); do
tmux new-window -t "$SESSION_NAME"
done
# 任务队列
TASKS=(
"审查 src/auth/ 目录的安全问题"
"重构 src/services/ 中的重复代码"
"为 src/api/ 补充单元测试"
"优化 src/db/ 中的数据库查询"
"检查 src/config/ 的配置一致性"
"分析 src/utils/ 的工具函数"
"更新 src/types/ 的类型定义"
"修复 src/handlers/ 的错误处理"
)
# 分配任务到窗口
task_idx=0
for window_idx in $(seq 0 $((MAX_WINDOWS - 1))); do
window_name="codex-$window_idx"
while [ $task_idx -lt ${#TASKS[@]} ]; do
# 在当前窗口执行任务
tmux send-keys -t "${SESSION_NAME}:${window_idx}" \
"cd /opt/data/my-project-wt-${window_idx} && \
codex exec --full-auto \"${TASKS[$task_idx]}\" && \
echo \"Task ${task_idx} done in window ${window_idx}\"" \
Enter
((task_idx++))
# 如果任务已分配完，跳出
if [ $task_idx -ge ${#TASKS[@]} ]; then
break
fi
# 给每个任务一个简短的间隔
sleep 1
done
done
echo "✅ Started $MAX_WINDOWS concurrent Codex sessions"
echo "📺 View: tmux attach -t $SESSION_NAME"
echo "📋 List: tmux list-windows -t $SESSION_NAME"三、CPU/内存监控
3.1 实时监控脚本
并发控制不只是"限制数量"，还需要"实时观察"系统状态：
#!/bin/bash
# codex-resource-monitor.sh —— Codex 进程资源监控
set -euo pipefail
# 监控阈值
CPU_WARN_THRESHOLD=80 # CPU 警告阈值（%）
CPU_CRITICAL_THRESHOLD=95 # CPU 临界阈值（%）
MEM_WARN_THRESHOLD=80 # 内存警告阈值（%）
MEM_CRITICAL_THRESHOLD=95 # 内存临界阈值（%）
# 获取所有 Codex 进程的资源使用
get_codex_stats() {
echo "=== Codex Process Resource Usage ==="
echo ""
# 查找所有 codex 进程
local pids=$(pgrep -f "codex exec" 2>/dev/null || true)
if [ -z "$pids" ]; then
echo "No Codex processes running."
return
fi
local total_cpu=0
local total_mem=0
local count=0
printf "%-10s %-8s %-8s %-8s %s\n" "PID" "CPU%" "MEM%" "RSS(MB)" "COMMAND"
printf "%-10s %-8s %-8s %-8s %s\n" "------" "----" "----" "-------" "-------"
for pid in $pids; do
if kill -0 "$pid" 2>/dev/null; then
local stats=$(ps -p "$pid" -o pid=,pcpu=,pmem=,rss=,args= 2>/dev/null || true)
if [ -n "$stats" ]; then
local cpu=$(echo "$stats" | awk '{print $2}')
local mem=$(echo "$stats" | awk '{print $3}')
local rss_kb=$(echo "$stats" | awk '{print $4}')
local rss_mb=$((rss_kb / 1024))
local cmd=$(echo "$stats" | awk '{for(i=5;i<=NF;i++) printf "%s ", $i; print ""}' | head -c 60)
printf "%-10s %-8s %-8s %-8s %s\n" "$pid" "$cpu" "$mem" "$rss_mb" "$cmd"
total_cpu=$(echo "$total_cpu + $cpu" | bc 2>/dev/null || echo "$total_cpu")
total_mem=$(echo "$total_mem + $mem" | bc 2>/dev/null || echo "$total_mem")
((count++))
fi
fi
done
echo ""
echo "Total: $count processes | CPU: ${total_cpu}% | MEM: ${total_mem}%"
echo ""
# 系统总体资源
local sys_cpu=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' 2>/dev/null || echo "N/A")
local sys_mem_total=$(free -m | awk '/Mem:/ {print $2}')
local sys_mem_used=$(free -m | awk '/Mem:/ {print $3}')
local sys_mem_pct=$((sys_mem_used * 100 / sys_mem_total))
echo "System: CPU=${sys_cpu}% | Memory=${sys_mem_used}MB/${sys_mem_total}MB (${sys_mem_pct}%)"
}
# 检查是否需要降级
check_degradation_needed() {
local sys_mem_pct=$(free | awk '/Mem:/ {printf "%.0f", $3/$2 * 100}')
local sys_cpu=$(top -bn1 | grep "Cpu(s)" | awk '{print $2 + $4}' 2>/dev/null || echo 0)
sys_cpu=${sys_cpu%.*} # 取整数部分
if [ "$sys_cpu" -ge "$CPU_CRITICAL_THRESHOLD" ] || [ "$sys_mem_pct" -ge "$MEM_CRITICAL_THRESHOLD" ]; then
echo "CRITICAL"
return 2
elif [ "$sys_cpu" -ge "$CPU_WARN_THRESHOLD" ] || [ "$sys_mem_pct" -ge "$MEM_WARN_THRESHOLD" ]; then
echo "WARNING"
return 1
else
echo "OK"
return 0
fi
}
# 持续监控模式
monitor_loop() {
local interval=${1:-5} # 默认 5 秒
echo "Starting resource monitor (interval: ${interval}s)"
echo "Press Ctrl+C to stop"
echo ""
while true; do
clear
get_codex_stats
echo ""
local status=$(check_degradation_needed)
case "$status" in
OK)
echo "🟢 System status: OK"
;;
WARNING)
echo "🟡 System status: WARNING - Consider reducing concurrency"
;;
CRITICAL)
echo "🔴 System status: CRITICAL - Initiate graceful degradation"
;;
esac
sleep "$interval"
done
}
# 入口
case "${1:-monitor}" in
stats)
get_codex_stats
;;
check)
check_degradation_needed
;;
monitor)
monitor_loop "${2:-5}"
;;
*)
echo "Usage: $0 {stats|check|monitor [interval]}"
exit 1
;;
esac3.2 cgroups 资源限制
如果你使用的是 Linux 系统，可以用 cgroups 对 Codex 进程组施加硬限制：
#!/bin/bash
# cgroup-codex-limit.sh —— 使用 cgroups 限制 Codex 资源
set -euo pipefail
CGROUP_NAME="codex-batch"
CPU_QUOTA="200000" # 2 个 CPU 核心（100000 = 1 core）
MEM_LIMIT="4G" # 4GB 内存上限
# 创建 cgroup（cgroup v2）
setup_cgroup() {
sudo mkdir -p /sys/fs/cgroup/"$CGROUP_NAME"
# 设置 CPU 配额
echo "$CPU_QUOTA" | sudo tee /sys/fs/cgroup/"$CGROUP_NAME"/cpu.max
# 设置内存限制
echo "$MEM_LIMIT" | sudo tee /sys/fs/cgroup/"$CGROUP_NAME"/memory.max
echo "✅ cgroup '$CGROUP_NAME' created"
echo " CPU quota: $((CPU_QUOTA / 100000)) cores"
echo " Memory limit: $MEM_LIMIT"
}
# 将进程加入 cgroup
add_to_cgroup() {
local pid=$1
sudo echo "$pid" | sudo tee /sys/fs/cgroup/"$CGROUP_NAME"/cgroup.procs
echo "Added PID $pid to cgroup '$CGROUP_NAME'"
}
# 清理 cgroup
cleanup_cgroup() {
sudo rmdir /sys/fs/cgroup/"$CGROUP_NAME" 2>/dev/null || true
echo "✅ cgroup '$CGROUP_NAME' removed"
}
# 在 cgroup 中运行 Codex
run_in_cgroup() {
local task=$1
# 使用 systemd-run 更方便（推荐）
sudo systemd-run \
--slice="$CGROUP_NAME" \
--property="CPUQuota=$((CPU_QUOTA / 1000))%" \
--property="MemoryMax=$MEM_LIMIT" \
--unit="codex-task-$$" \
codex exec --full-auto "$task"
}
# 入口
case "${1:-}" in
setup)
setup_cgroup
;;
run)
run_in_cgroup "$2"
;;
cleanup)
cleanup_cgroup
;;
*)
echo "Usage: $0 {setup|run <task>|cleanup}"
;;
esac3.3 结合 prometheus 的长期监控
对于生产环境，建议接入 Prometheus + Grafana 做长期监控：
# prometheus-codex.yml —— Prometheus 自定义指标采集
scrape_configs:
- job_name: 'codex-concurrency'
static_configs:
- targets: ['localhost:9090']
metrics_path: '/metrics'
# 自定义 exporter 采集的指标：
# codex_running_processes - 当前运行的 Codex 进程数
# codex_cpu_usage_percent - Codex 进程组 CPU 使用率
# codex_memory_usage_bytes - Codex 进程组内存使用
# codex_task_duration_seconds - 任务执行时长
# codex_task_failures_total - 任务失败次数
# codex_api_rate_limit_remaining - API 剩余配额四、优雅降级
4.1 降级策略设计
优雅降级的核心思想：当系统资源不足时，主动降低负载，而不是被动崩溃。
┌─────────────────────────────────────────────────────────┐
│ 优雅降级策略矩阵 │
├─────────────┬──────────────┬──────────────┬─────────────┤
│ 级别 │ 触发条件 │ 动作 │ 恢复条件 │
├─────────────┼──────────────┼──────────────┼─────────────┤
│ GREEN │ CPU<70% │ 正常并发 │ - │
│ │ MEM<70% │ (4 实例) │ │
├─────────────┼──────────────┼──────────────┼─────────────┤
│ YELLOW │ CPU 70-85% │ 减少到 2 实例 │ CPU<60% │
│ │ MEM 70-85% │ 暂停新任务 │ MEM<60% │
│ │ │ 10 秒冷却 │ 持续 30 秒 │
├─────────────┼──────────────┼──────────────┼─────────────┤
│ ORANGE │ CPU 85-95% │ 减少到 1 实例 │ CPU<70% │
│ │ MEM 85-95% │ 等待完成中 │ MEM<70% │
│ │ │ 不启动新的 │ 持续 60 秒 │
├─────────────┼──────────────┼──────────────┼─────────────┤
│ RED │ CPU>95% │ 全部暂停 │ CPU<80% │
│ │ MEM>95% │ 发送告警 │ MEM<80% │
│ │ │ 保留现场 │ 持续 120 秒 │
└─────────────┴──────────────┴──────────────┴─────────────┘4.2 完整的降级控制器
#!/bin/bash
# codex-degradation-controller.sh —— 优雅降级控制器
set -euo pipefail
# 状态文件
STATE_FILE="/tmp/codex-degradation-state.json"
LOCK_FILE="/tmp/codex-degradation.lock"
# 降级级别
LEVEL_GREEN=0
LEVEL_YELLOW=1
LEVEL_ORANGE=2
LEVEL_RED=3
# 最大并发数对应每个级别
declare -A MAX_CONCURRENT_BY_LEVEL
MAX_CONCURRENT_BY_LEVEL[$LEVEL_GREEN]=4
MAX_CONCURRENT_BY_LEVEL[$LEVEL_YELLOW]=2
MAX_CONCURRENT_BY_LEVEL[$LEVEL_ORANGE]=1
MAX_CONCURRENT_BY_LEVEL[$LEVEL_RED]=0
# 获取当前系统状态
get_system_metrics() {
local cpu=$(top -bn1 | grep "Cpu(s)" | awk '{print $2 + $4}' 2>/dev/null || echo 0)
cpu=${cpu%.*}
local mem_pct=$(free | awk '/Mem:/ {printf "%.0f", $3/$2 * 100}')
local load_1m=$(cat /proc/loadavg | awk '{print $1}')
echo "{\"cpu\":$cpu,\"mem_pct\":$mem_pct,\"load_1m\":$load_1m}"
}
# 计算需要的降级级别
calculate_level() {
local cpu=$1
local mem_pct=$2
local load_1m=$3
local cpu_cores=$(nproc)
# CPU 判断
if [ "$cpu" -ge 95 ] || [ "$mem_pct" -ge 95 ]; then
echo $LEVEL_RED
return
elif [ "$cpu" -ge 85 ] || [ "$mem_pct" -ge 85 ]; then
echo $LEVEL_ORANGE
return
elif [ "$cpu" -ge 70 ] || [ "$mem_pct" -ge 70 ]; then
echo $LEVEL_YELLOW
return
else
echo $LEVEL_GREEN
return
fi
}
# 执行降级动作
apply_degradation() {
local new_level=$1
local old_level=$2
local max_concurrent=${MAX_CONCURRENT_BY_LEVEL[$new_level]}
echo "[$(date '+%Y-%m-%d %H:%M:%S')] Degradation: $old_level → $new_level (max concurrent: $max_concurrent)"
# 更新状态文件
local metrics=$(get_system_metrics)
echo "{\"level\":$new_level,\"max_concurrent\":$max_concurrent,\"metrics\":$metrics,\"timestamp\":\"$(date -Iseconds)\"}" \
> "$STATE_FILE"
case $new_level in
$LEVEL_RED)
# 红色：停止所有新任务
touch "$LOCK_FILE"
echo "🔴 CRITICAL: All new Codex tasks paused. Send alert."
# 发送告警（集成钉钉、企业微信、Slack 等）
send_alert "CRITICAL" "Codex degradation to RED level"
;;
$LEVEL_ORANGE)
# 橙色：只允许 1 个并发
echo "🟠 WARNING: Reducing to 1 concurrent instance"
;;
$LEVEL_YELLOW)
# 黄色：减少到 2 个并发
echo "🟡 NOTICE: Reducing to 2 concurrent instances"
;;
$LEVEL_GREEN)
# 绿色：恢复正常
rm -f "$LOCK_FILE"
echo "🟢 OK: Full concurrency restored"
;;
esac
}
# 发送告警（示例：curl 调用 Webhook）
send_alert() {
local level=$1
local message=$2
# 企业微信 Webhook 示例
# curl -s "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY" \
# -H "Content-Type: application/json" \
# -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"[${level}] Codex: ${message}\"}}"
# 钉钉 Webhook 示例
# curl -s "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN" \
# -H "Content-Type: application/json" \
# -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"[${level}] Codex: ${message}\"}}"
echo "Alert sent: [$level] $message"
}
# 降级监控循环（守护进程模式）
degradation_daemon() {
local check_interval=${1:-5} # 默认 5 秒检查一次
local recovery_delay=${2:-30} # 恢复延迟 30 秒
echo "Starting degradation controller daemon"
echo "Check interval: ${check_interval}s | Recovery delay: ${recovery_delay}s"
local current_level=$LEVEL_GREEN
local consecutive_ok=0
while true; do
local metrics=$(get_system_metrics)
local cpu=$(echo "$metrics" | jq -r '.cpu')
local mem_pct=$(echo "$metrics" | jq -r '.mem_pct')
local load_1m=$(echo "$metrics" | jq -r '.load_1m')
local needed_level=$(calculate_level "$cpu" "$mem_pct" "$load_1m")
# 防止抖动：需要连续 OK 才恢复
if [ "$needed_level" -eq "$LEVEL_GREEN" ] && [ "$current_level" -gt "$LEVEL_GREEN" ]; then
((consecutive_ok++))
if [ "$consecutive_ok" -ge "$((recovery_delay / check_interval))" ]; then
apply_degradation $LEVEL_GREEN $current_level
current_level=$LEVEL_GREEN
consecutive_ok=0
fi
else
consecutive_ok=0
if [ "$needed_level" -ne "$current_level" ]; then
apply_degradation "$needed_level" "$current_level"
current_level=$needed_level
fi
fi
sleep "$check_interval"
done
}
# 入口
case "${1:-daemon}" in
daemon)
degradation_daemon "${2:-5}" "${3:-30}"
;;
check)
get_system_metrics
;;
status)
if [ -f "$STATE_FILE" ]; then
jq . "$STATE_FILE"
else
echo "No degradation state found. System running normally."
fi
;;
*)
echo "Usage: $0 {daemon [check_interval] [recovery_delay]|check|status}"
;;
esac4.3 任务优先级队列
降级时，我们还需要对任务进行优先级排序：
#!/bin/bash
# priority-queue-codex.sh —— 优先级任务队列
set -euo pipefail
# 优先级定义（数字越小优先级越高）
# 1 - CRITICAL: 安全漏洞修复、生产事故
# 2 - HIGH: 线上 Bug 修复
# 3 - MEDIUM: 功能开发、PR Review
# 4 - LOW: 文档更新、代码格式化
# 5 - BACKGROUND: 批量重构、技术债清理
PRIORITY_QUEUE=()
add_task() {
local priority=$1
local description=$2
local worktree=$3
# 按优先级插入（简单冒泡排序）
PRIORITY_QUEUE+=("${priority}|${description}|${worktree}")
# 重新排序
IFS=$'\n' PRIORITY_QUEUE=($(sort -t'|' -k1 -n <<< "${PRIORITY_QUEUE[*]}"))
unset IFS
}
process_priority_queue() {
local max_concurrent=$1
while [ ${#PRIORITY_QUEUE[@]} -gt 0 ]; do
# 取最高优先级的任务
local task="${PRIORITY_QUEUE[0]}"
PRIORITY_QUEUE=("${PRIORITY_QUEUE[@]:1}")
IFS='|' read -r priority description worktree <<< "$task"
echo "Processing [Priority $priority]: $description"
echo " Worktree: $worktree"
cd "$worktree"
codex exec --full-auto "$description"
echo " ✅ Complete"
echo ""
done
}
# 示例
add_task 3 "Review PR #42: 新增用户注册 API" "/tmp/codex-wt-pr42"
add_task 1 "修复生产环境 SQL 注入漏洞" "/tmp/codex-wt-hotfix"
add_task 5 "批量更新所有文件的 License Header" "/tmp/codex-wt-license"
add_task 2 "修复登录超时问题" "/tmp/codex-wt-login-fix"
add_task 4 "更新 README 文档" "/tmp/codex-wt-docs"
echo "Queue order (by priority):"
for task in "${PRIORITY_QUEUE[@]}"; do
IFS='|' read -r priority description _ <<< "$task"
echo " [$priority] $description"
done
echo ""
process_priority_queue 2五、实战：完整的并发管理系统
将上述所有模块整合为一个完整的管理系统：
#!/bin/bash
# codex-concurrency-manager.sh —— 完整的并发管理系统
set -euo pipefail
# ===== 配置 =====
MAX_CONCURRENT=${CODEX_MAX_CONCURRENT:-4}
CHECK_INTERVAL=5
CPU_WARN=75
CPU_CRIT=90
MEM_WARN=75
MEM_CRIT=90
RECOVERY_SECONDS=60
WORKTREE_BASE="/tmp/codex-wt"
LOG_DIR="/tmp/codex-logs"
STATE_DIR="/tmp/codex-state"
mkdir -p "$LOG_DIR" "$STATE_DIR" "$WORKTREE_BASE"
# ===== 日志 =====
log() { echo "[$(date '+%H:%M:%S')] [$1] $2"; }
# ===== 资源检查 =====
get_cpu_usage() {
top -bn1 | grep "Cpu(s)" | awk '{print $2 + $4}' 2>/dev/null | cut -d. -f1
}
get_mem_usage() {
free | awk '/Mem:/ {printf "%.0f", $3/$2 * 100}'
}
get_running_count() {
pgrep -f "codex exec" 2>/dev/null | wc -l
}
# ===== 降级判断 =====
check_status() {
local cpu=$(get_cpu_usage)
local mem=$(get_mem_usage)
local running=$(get_running_count)
if [ "$cpu" -ge "$CPU_CRIT" ] || [ "$mem" -ge "$MEM_CRIT" ]; then
echo "RED|0"
elif [ "$cpu" -ge "$CPU_WARN" ] || [ "$mem" -ge "$MEM_WARN" ]; then
local reduce=$((MAX_CONCURRENT / 2))
[ "$reduce" -lt 1 ] && reduce=1
echo "ORANGE|$reduce"
elif [ "$running" -ge "$MAX_CONCURRENT" ]; then
echo "YELLOW|$MAX_CONCURRENT"
else
echo "GREEN|$MAX_CONCURRENT"
fi
}
# ===== 任务调度 =====
declare -a TASK_QUEUE=()
declare -a RUNNING_PIDS=()
add_to_queue() {
TASK_QUEUE+=("$1")
log "QUEUE" "Added task: $1"
}
can_start_task() {
local status=$(check_status)
local level=$(echo "$status" | cut -d'|' -f1)
local allowed=$(echo "$status" | cut -d'|' -f2)
local running=$(get_running_count)
[ "$level" = "RED" ] && return 1
[ "$running" -ge "$allowed" ] && return 1
return 0
}
start_task() {
local task=$1
local task_id=$(date +%s%N | cut -c10-16)
local log_file="${LOG_DIR}/task-${task_id}.log"
(
echo "[$(date '+%Y-%m-%d %H:%M:%S')] Starting: $task" > "$log_file"
cd /opt/data/my-project 2>/dev/null || cd "$HOME"
codex exec --full-auto "$task" >> "$log_file" 2>&1
local rc=$?
echo "[$(date '+%Y-%m-%d %H:%M:%S')] Finished (exit: $rc): $task" >> "$log_file"
echo "$rc" > "${STATE_DIR}/task-${task_id}.exit"
) &
RUNNING_PIDS+=($!)
log "START" "Task $task_id started (PID: ${RUNNING_PIDS[-1]})"
}
cleanup_finished() {
local new_pids=()
for pid in "${RUNNING_PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
new_pids+=("$pid")
else
wait "$pid" 2>/dev/null
log "DONE" "PID $pid finished"
fi
done
RUNNING_PIDS=("${new_pids[@]+"${new_pids[@]}"}")
}
# ===== 调度循环 =====
scheduler_loop() {
log "SCHEDULER" "Starting scheduler loop"
while true; do
cleanup_finished
# 检查队列
if [ ${#TASK_QUEUE[@]} -gt 0 ]; then
if can_start_task; then
local task="${TASK_QUEUE[0]}"
TASK_QUEUE=("${TASK_QUEUE[@]:1}")
start_task "$task"
else
log "SCHEDULER" "Queue: ${#TASK_QUEUE[@]} pending, waiting for slot..."
fi
fi
local status=$(check_status)
local level=$(echo "$status" | cut -d'|' -f1)
local allowed=$(echo "$status" | cut -d'|' -f2)
local running=$(get_running_count)
log "STATUS" "Level=$level | Allowed=$allowed | Running=$running | Queue=${#TASK_QUEUE[@]}"
# 如果队列空且无运行中任务，退出
if [ ${#TASK_QUEUE[@]} -eq 0 ] && [ ${#RUNNING_PIDS[@]} -eq 0 ]; then
log "SCHEDULER" "All tasks complete"
break
fi
sleep "$CHECK_INTERVAL"
done
# 等待所有剩余任务
wait
log "SCHEDULER" "Scheduler loop exited"
}
# ===== 入口 =====
case "${1:-help}" in
add)
add_to_queue "$2"
;;
run)
scheduler_loop
;;
status)
echo "Running: $(get_running_count)"
echo "CPU: $(get_cpu_usage)%"
echo "MEM: $(get_mem_usage)%"
echo "Queue: ${#TASK_QUEUE[@]} tasks"
;;
help|*)
echo "Usage: $0 {add <task>|run|status|help}"
;;
esac总结
本文系统地介绍了 Codex CLI 多进程并发控制的完整方案：
- 进程数限制：固定上限、动态基于 CPU 核心数、GNU parallel、tmux 编排四种方式，从简单到复杂，适配不同场景
- CPU/内存监控：实时监控脚本、cgroups 硬限制、Prometheus 长期监控三级方案，确保你能随时掌握系统状态
- 优雅降级：四级降级策略（GREEN/YELLOW/ORANGE/RED）配合自动恢复机制，防止系统崩溃
- 优先级队列：在资源紧张时优先处理关键任务
- 完整管理系统：将上述模块整合为统一的调度器
核心原则：
- 限制永远比崩溃好：宁可慢一点，也不要让系统失控
- 监控必须先行：没有监控的并发控制是盲目的
- 降级要优雅：自动恢复 + 告警通知，让系统有自我修复能力
- 优先级是保险：关键任务永远优先获得资源
📌 下篇预告
成本控制 —— Token 预算管理、--yolo vs 沙箱模式的成本对比、模型选择策略（o3-mini vs o3 vs o4-mini 的性价比分析）、每日/每周/每月的成本监控和告警。
当你的团队每天运行数十个 Codex 任务时，成本控制就不再是一个可选项，而是必须面对的工程问题。下一篇我们将深入探讨如何用最小的预算，获得最大的产出。