Christian Emmer 分享在 Bash 中实现并发受限的命令队列

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（zsh 的重定向上限）中找出第一个未使用的 fd 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 分隔的命令。通过关闭作业控制并丢弃该代码块的 35 # stderr 来隐藏作业消息。xargs 通过队列文件 fd 上的一份副本拿到真正的 stderr，直到该文件被打开。 36 eval "exec ${fd}>&2" 37 local opts=$- 38 set +m 39 { 40 # shellcheck disable=SC2016 41 xargs -0 -n 1 -P "$1" "${shell}" -c ' 42 # 输出被捕获而非流式转发，以便各命令的输出保持分离 43 out=$(eval -- "$1" </dev/null 2>&1); rc=$? 44 [ -z "${out}" ] || printf "%s\n" "${out}" 45 # 把非零退出码强制为 1，因为 255 会让 xargs 停止 46 [ "${rc}" -eq 0 ] || { printf "exit %d: %s\n" "${rc}" "$1" >&2; exit 1; 47 }' _ <"${dir}/queue" 2>&"${fd}" & 48 _queue_pid=$! 49 disown || : 50 } 2>/dev/null 51 if [[ ${opts} == m ]]; then set -m; fi 52 53 # 以写入方式打开队列文件，这会等到 xargs 以读取方式打开它 54 # 两端都打开后，就可以安全地删除该文件 55 eval "exec ${fd}>"${dir}/queue"" 56 _queue_fd=${fd} 57 rm -rf "${dir}" 58 59 # 注册退出处理器，终止本 PID 的子进程，确保 zsh 的行为与 bash 一致 60 setopt localoptions posixtraps 2>/dev/null || : 61 trap _queue_stop EXIT 62 if [[ $- != i ]]; then 63 trap 'exit 129' HUP 64 trap 'exit 130' INT 65 trap 'exit 143' TERM 66 fi 67} 68 69# 向当前 shell 的命令队列提交一条命令。 70# @param {string} $1 要运行的命令 71queue_submit() { 72 if [[ -z ${_queue_fd:-} ]]; then 73 echo "queue_submit: no command queue started in this shell" >&2 74 return 1 75 fi 76 # 把命令写入队列文件，若队列已满则阻塞 77 # 使用一个 printf 进程，这样若 xargs 已退出，被杀的会是 printf 而不是当前 shell 78 if [[ $- == i ]]; then 79 # 在前台写入命令，因为交互式 shell 没有 TERM/HUP/INT trap， 80 # 不会因队列满而被延迟 81 env printf '%s\0' "$1" >&"${_queue_fd}" 82 else 83 # 在后台写入命令，这样队列满不会延迟任何 TERM/HUP/INT trap 84 env printf '%s\0' "$1" >&"${_queue_fd}" & 85 wait "$!" 86 fi || { 87 echo "queue_submit: couldn't queue the command; the command queue may have stopped" >&2 88 return 1 89 } 90} 91 92# 关闭队列文件，让 xargs 执行完手头的作业，然后等待它退出 93queue_wait() { 94 if [[ -z ${_queue_fd:-} ]]; then 95 echo "queue_wait: no command queue started in this shell" >&2 96 return 1 97 fi 98 eval "exec ${_queue_fd}>&-" 99 while kill -0 "${_queue_pid}" 2>/dev/null; do sleep 0.1; done 100 unset _queue_fd _queue_pid 101} 102 103# 停止给定 PID 下的所有进程。 104# @param {number=} $1 要停止其后代的 PID，默认为当前 shell 的 PID 105_queue_stop() { 106 # 默认使用当前 shell 的 PID，从子进程的父 PID 获取，因为在子 shell 中 $$ 是父进程的 PID 107 local parent=${1:-$(exec sh -c 'echo "${PPID}"')} 108 local pid 109 for pid in $(exec pgrep -P "${parent}"); do 110 # 暂停进程，使其无法再创建新子进程 111 kill -STOP "${pid}" 2>/dev/null || continue 112 # 终止该进程的后代，然后终止该进程 113 _queue_stop "${pid}" 114 kill -TERM "${pid}" 2>/dev/null || : 115 # 恢复进程，使其能响应终止信号 116 kill -CONT "${pid}" 2>/dev/null || : 117 done 118} 119 120# 退出时停止所有后台进程，因为它们忽略 Ctrl-C 且收不到 TERM 121if [[ $- != i ]]; then 122 trap _queue_stop EXIT 123 trap 'exit 129' HUP 124 trap 'exit 130' INT 125 trap 'exit 143' TERM 126fi queue_start 初始化队列（由 mkfifo 管理）并启动执行器（并发数由 xargs -P 管理），queue_submit 向队列提交命令，queue_wait 等待队列排空后再清理。调用 queue_wait 是可选的，命令一旦通过 queue_submit 入队，处理就会开始。

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