用 Bash 实现并发受限的 Shell 命令队列

Christian Emmer 2026年11月6日* · 阅读时长10分钟 如何并行运行 shell 命令，同时限制并发数。 我最近遇到一个场景：想加快一些 Bash 处理，但又要把并发数控制住。于是我就想：Bash 里有没有直接的办法做线程池？答案是：有，但要做得半可靠、可移植，还得费点功夫。 你可能会问：为什么非要用 Bash 做这件事？用一个并发控制很弱的语言，看着挺傻的。这问题问得对，我的回答是：因为我喜欢 Bash，而且我想挑战一下。不过说实话，换一门别的语言，比如 Python，你大概率会更省事。 下面这段代码创建了一个命令队列，用来接收要执行的 shell 命令。你不需要事先知道所有想执行的命令，可以随时间陆续提交。 1# 启动当前 shell 的命令队列。 2# @param {number} $1 同时运行的最大任务数 3queue_start() { 4 # 校验并行度是否是 xargs 能接受的有效上限 5 if [[ ! ${1:-} =~ ^[1-9][0-9]*$ ]]; then 6 echo "queue_start: size must be a positive integer, got '${1:-}'" >&2 7 return 1 8 fi 9 10 # 关闭当前 shell（不是父 shell）对继承来的或更早的命令队列文件描述符的副本 11 if [[ -n ${_queue_fd:-} ]]; then 12 eval "exec ${_queue_fd}>&-" 13 unset _queue_fd _queue_pid 14 fi 15 16 # 找出 3 到 9 之间第一个未使用的 fd（zsh 重定向的上限） 17 local fd=3 18 while { : <&"${fd}"; } 2>/dev/null; do fd=$(( fd + 1 )); done 19 if [[ ${fd} -gt 9 ]]; then 20 echo "queue_start: no free fd from 3 to 9" >&2 21 return 1 22 fi 23 24 # 在一个私有的临时目录里创建 FIFO 队列文件 25 local dir 26 dir=$(mktemp -d "${TMPDIR:-/tmp}/queue.XXXXXX") 27 mkfifo "${dir}/queue" 28 29 # 找出用于执行命令的 shell（即当前 shell） 30 local shell 31 shell=$(ps -o comm= -p "$$") 32 shell=${shell#-} 33 34 # 在后台启动 xargs，从队列文件读取以 NUL 分隔的命令。通过关闭作业控制并丢弃该代码块的 stderr 来隐藏作业消息。xargs 通过队列文件 fd 上的一个副本获得真正的 stderr，直到该文件被打开。 35 eval "exec ${fd}>&2" 36 local opts=$- 37 set +m 38 { 39 # shellcheck disable=SC2016 40 xargs -0 -n 1 -P "$1" "${shell}" -c ' 41 # 输出是捕获后再打印的，而不是直接流式输出，这样各条命令的输出不会混在一起 42 out=$(eval -- "$1" </dev/null 2>&1); rc=$? 43 [ -z "${out}" ] || printf "%s\n" "${out}" 44 # 把非零退出码统一改成 1，因为 255 会让 xargs 停止 45 [ "${rc}" -eq 0 ] || { printf "exit %d: %s\n" "${rc}" "$1" >&2; exit 1; 46 }' _ <"${dir}/queue" 2>&"${fd}" & 47 _queue_pid=$! 48 disown || : 49 } 2>/dev/null 50 if [[ ${opts} == m ]]; then set -m; fi 51 52 # 以写方式打开队列文件，这一步会等到 xargs 以读方式打开它 53 # 两端都打开后，文件就可以安全删除了 54 eval "exec ${fd}>"${dir}/queue"" 55 _queue_fd=${fd} 56 rm -rf "${dir}" 57 58 # 注册退出处理器，用来终止本 PID 的子进程，确保 zsh 的行为与 bash 一致 59 setopt localoptions posixtraps 2>/dev/null || : 60 trap _queue_stop EXIT 61 if [[ $- != i ]]; then 62 trap 'exit 129' HUP 63 trap 'exit 130' INT 64 trap 'exit 143' TERM 65 fi 66} 67 68# 在当前 shell 的命令队列上排入一条命令。 69# @param {string} $1 要运行的命令 70queue_submit() { 71 if [[ -z ${_queue_fd:-} ]]; then 72 echo "queue_submit: no command queue started in this shell" >&2 73 return 1 74 fi 75 # 把命令写入队列文件，如果队列已满则阻塞 76 # 用一个 printf 进程来写，这样即使 xargs 已退出，被杀掉的也是 printf 而不是当前 shell 77 if [[ $- == i ]]; then 78 # 在前台写入命令，因为交互式 shell 不会有任何可能被满队列延迟的 TERM/HUP/INT trap 79 env printf '%s\0' "$1" >&"${_queue_fd}" 80 else 81 # 在后台写入命令，这样队列满时不会延迟任何 TERM/HUP/INT trap 82 env printf '%s\0' "$1" >&"${_queue_fd}" & 83 wait "$!" 84 fi || { 85 echo "queue_submit: couldn't queue the command; the command queue may have stopped" >&2 86 return 1 87 } 88} 89 90# 关闭队列文件，让 xargs 执行完手上的任务，然后等待它退出 91queue_wait() { 92 if [[ -z ${_queue_fd:-} ]]; then 93 echo "queue_wait: no command queue started in this shell" >&2 94 return 1 95 fi 96 eval "exec ${_queue_fd}>&-" 97 while kill -0 "${_queue_pid}" 2>/dev/null; do sleep 0.1; done 98 unset _queue_fd _queue_pid 99} 100 101# 停止给定 PID 下的所有进程。 102# @param {number=} $1 要停止其后代的 PID，默认为当前 shell 的 PID 103_queue_stop() { 104 # 默认取当前 shell 的 PID；在子 shell 中 $$ 是父进程的 PID，因此从子进程的父 PID 获取 105 local parent=${1:-$(exec sh -c 'echo "${PPID}"')} 106 local pid 107 for pid in $(exec pgrep -P "${parent}"); do 108 # 先暂停进程，使其无法再启动新的子进程 109 kill -STOP "${pid}" 2>/dev/null || continue 110 # 先终止该进程的后代，再终止该进程 111 _queue_stop "${pid}" 112 kill -TERM "${pid}" 2>/dev/null || : 113 # 恢复进程，让它能响应终止信号 114 kill -CONT "${pid}" 2>/dev/null || : 115 done 116} 117 118# 退出时停止所有后台进程，因为它们会忽略 Ctrl-C，也收不到 TERM 119if [[ $- != i ]]; then 120 trap _queue_stop EXIT 121 trap 'exit 129' HUP 122 trap 'exit 130' INT 123 trap 'exit 143' TERM 124fi queue_start 负责初始化队列（由 mkfifo 管理）并启动执行器（并发数由 xargs -P 控制），queue_submit 把命令提交到队列，queue_wait 则等待队列排空后再做清理。queue_wait 的调用是可选的，命令一旦通过 queue_submit 排入队列，处理就会立即开始。

（原文较长，此处译出前 6000 字，完整内容请点「打开原文」。）