Newshot 并发 Shell 命令队列
如何并行运行 Shell 命令并限制其并发数。
我最近有一个用例，我想加速一些我正在做的 Bash 处理，但仍然保持并发受限。所以我想知道：在 Bash 中创建线程池有没有一种直接的方法？答案是：有，但需要一些工作才能做到半可靠和可移植。
你可能会问：为什么要在 Bash 中做这件事？使用一个并发控制很差的语音似乎很傻。你问这个问题是对的，我的回答是：因为我对 Bash 有感情，而且我想要挑战。但你可能仍然最好使用另一种语言，比如 Python。
下面的代码创建了一个命令队列，可以接受要执行的 shell 命令。你不需要预先知道所有想要执行的命令，你可以随时间向它提交命令。
1# Start this shell's command queue.
2# @param {number} $1 Maximum number of jobs to run at once
3queue_start() {
4 # Validate that the parallelism is a valid limit for 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 # Close this shell's copy (not the parent's) of an inherited or earlier
11 # command queue's file descriptor
12 if [[ -n ${_queue_fd:-} ]]; then
13 eval "exec ${_queue_fd}>&-"
14 unset _queue_fd _queue_pid
15 fi
16
17 # Find the first unused fd from 3 to 9 (zsh's redirect max)
18 local fd=3
19 while { : <&"${fd}"; } 2>/dev/null; do fd=$(( fd + 1 )); done
20 if [[ ${fd} -gt 9 ]]; then
21 echo "queue_start: no free fd from 3 to 9" >&2
22 return 1
23 fi
24
25 # Make a FIFO queue file in a private temporary directory
26 local dir
27 dir=$(mktemp -d "${TMPDIR:-/tmp}/queue.XXXXXX")
28 mkfifo "${dir}/queue"
29
30 # Find the shell to execute commands (the current shell)
31 local shell
32 shell=$(ps -o comm= -p "$$")
33 shell=${shell#-}
34
35 # Start xargs in the background, reading NUL-separated commands from the
36 # queue file. Job messages are hidden by turning off job control and
37 # discarding the block's stderr. xargs gets the real stderr through a copy on
38 # the queue files's fd, until the file is opened.
39 eval "exec ${fd}>&2"
40 local opts=$-
41 set +m
42 {
43 # shellcheck disable=SC2016
44 xargs -0 -n 1 -P "$1" "${shell}" -c '
45 # Output is captured rather than streamed so that command output stays separate
46 out=$(eval -- "$1" </dev/null 2>&1); rc=$?
47 [ -z "${out}" ] || printf "%s\n" "${out}"
48 # Coerce non-zero exit codes to 1, because a 255 code would stop xargs
49 [ "${rc}" -eq 0 ] || { printf "exit %d: %s\n" "${rc}" "$1" >&2; exit 1;
50 }' _ <"${dir}/queue" 2>&"${fd}" &
51 _queue_pid=$!
52 disown || :
53 } 2>/dev/null
54 if [[ ${opts} == *m* ]]; then set -m; fi
55
56 # Open the queue file for writing, which waits until xargs has opened it for reading
57 # With both ends open, the file can be safely deleted
58 eval "exec ${fd}>\"\${dir}/queue\""
59 _queue_fd=${fd}
60 rm -rf "${dir}"
61
62 # Register an exit handler to terminate this PID's children, ensuring zsh behaves like bash
63 setopt localoptions posixtraps 2>/dev/null || :
64 trap _queue_stop EXIT
65 if [[ $- != *i* ]]; then
66 trap 'exit 129' HUP
67 trap 'exit 130' INT
68 trap 'exit 143' TERM
69 fi
70}
71
72# 在当前 shell 的命令队列中排队一条命令。
73# @param {string} $1 要运行的命令
74queue_submit() {
75 if [[ -z ${_queue_fd:-} ]]; then
76 echo "queue_submit: no command queue started in this shell" >&2
77 return 1
78
$ queue_start 2 $ ( queue_submit 'echo "submitted from a subshell"' ) $ queue_submit 'echo "submitted from the main shell"' $ queue_wait submitted from a subshell submitted from the main shell
不同的子 shell 可以拥有自己的命令队列：
$ ( queue_start 2 for n in 1 2 3 4; do queue_submit "sleep 1; echo first queue, job ${n}"; done queue_wait ) & $ first=$! $ ( queue_start 1 for n in 1 2; do queue_submit "sleep 1; echo second queue, job ${n}"; done queue_wait ) & $ second=$! $ wait "${first}" $ wait "${second}" second queue, job 1 first queue, job 1 first queue, job 2 second queue, job 2 first queue, job 3 first queue, job 4
将命令输出的每一行作为待执行命令加入队列：
$ queue_start 2 $ while IFS= read -r line; do queue_submit "${line}" done < <(printf 'echo "from command output, line %d"\n' 1 2 3) $ queue_wait from command output, line 1 from command output, line 2 from command output, line 3
将文件的每一行作为命令加入队列：
$ queue_start 2 $ while IFS= read -r line; do queue_submit "${line}" done < commands.txt $ queue_wait
## 已知问题
以下是一些你可能想换用其他编程语言的理由：
- 一个 shell 只能有一个队列。支持多个队列所需的额外代码很复杂，而且在大多数情况下很可能并不需要。子 shell 仍然可以拥有自己的队列。
一个 shell 只能有一个队列。
支持多个队列所需的额外代码很复杂，而且在大多数情况下很可能并不需要。子 shell 仍然可以拥有自己的队列。
- 命令输出有意不进行流式传输。这可以防止大多数命令的输出交错。大于 PIPE_BUF 的输出仍可能交错。
命令输出有意不进行流式传输。
这可以防止大多数命令的输出交错。大于 PIPE_BUF 的输出仍可能交错。
- queue_wait 不返回表示任何作业失败的值。**这需要临时文件或第二个文件描述符，以及一些复杂的代码。
queue_wait 不返回表示任何作业失败的值。**
这需要临时文件或第二个文件描述符，以及一些复杂的代码。
- queue_wait 必须等待 queue_start 之后启动的运行中进程。在 queue_start 之后启动的进程会继承队列写入文件描述符（这正是子 shell 能够工作的原因），并且在写入描述符的每个副本都关闭之前，xargs 不会看到队列的结尾。
queue_wait 必须等待 queue_start 之后启动的运行中进程。
在 queue_start 之后启动的进程会继承队列写入文件描述符（这正是子 shell 能够工作的原因），并且在写入描述符的每个副本都关闭之前，xargs 不会看到队列的结尾。
- _queue_stop 会不加区分地终止后代进程。任何其他后台作业都会被终止，包括在 queue_start 之前启动的作业。
_queue_stop 会不加区分地终止后代进程。
任何其他后台作业都会被终止，包括在 queue_start 之前启动的作业。
## 为什么不直接用 xargs -P？
如果你事先知道所有想要执行的命令，那么解决方案会简单得多：
$ printf 'echo "from command output, line %d"\0' 1 2 3 | \ xargs -0 -n 1 -P 2 bash -c from command output, line 1 from command output, line 2 from command output, line 3
$ tr '\n' '\0' < commands.txt | \ xargs -0 -n 1 -P 2 bash -c
GNU parallel 的工作方式类似，不过 macOS 并不自带它：
$ printf 'echo "from command output, line %d"\n' 1 2 3 | \ parallel -j 2 from command output, line 1 from command output, line 2 from command output, line 3