bash - 在 bash 脚本中并行运行数组中的每个元素

Question

假设我有一个如下所示的 bash 脚本:

array=( 1 2 3 4 5 6 )

for each in "${array[@]}"
do
  echo "$each"

  command --arg1 $each

done

如果我想并行运行循环中的所有内容，我可以将 command --arg1 $each 更改为 command --arg1 $each &。

但现在假设我想获取 command --arg1 $each 的结果，并对这些结果执行如下操作:

array=( 1 2 3 4 5 6 )
for each in "${array[@]}"
do
  echo "$each"

  lags=($(command --arg1 $each)

  lngth_lags=${#lags[*]}

  for (( i=1; i<=$(( $lngth_lags -1 )); i++))
  do

    result=${lags[$i]}
    echo -e "$timestamp\t$result" >> $log_file
    echo "result piped"

  done

done

如果我只是在 command --arg1 $each 的末尾添加一个 &，command --arg1 $each 之后的所有内容都会在没有 command --arg1 $each 首先完成的情况下运行。我如何防止这种情况发生？另外，如何限制循环可以占用的线程数？

本质上，这个 block 应该并行运行 1,2,3,4,5,6

  echo "$each"

  lags=($(command --arg1 $each)

  lngth_lags=${#lags[*]}

  for (( i=1; i<=$(( $lngth_lags -1 )); i++))
  do

    result=${lags[$i]}
    echo -e "$timestamp\t$result" >> $log_file
    echo "result piped"

  done

-----编辑--------

原代码如下:

#!/bin/bash
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/etc/kafka/kafka.client.jaas.conf"
IFS=$'\n'
array=($(kafka-consumer-groups --bootstrap-server kafka1:9092 --list --command-config /etc/kafka/client.properties --new-consumer))

lngth=${#array[*]}

echo "array length: " $lngth

timestamp=$(($(date +%s%N)/1000000))

log_time=`date +%Y-%m-%d:%H`

echo "log time: " $log_time

log_file="/home/ec2-user/laglogs/laglog.$log_time.log"

echo "log file: " $log_file

echo "timestamp: " $timestamp

get_lags () {

  echo "$1"

  lags=($(kafka-consumer-groups --bootstrap-server kafka1:9092 --describe  --group $1 --command-config /etc/kafka/client.properties --new-consumer))

  lngth_lags=${#lags[*]}

  for (( i=1; i<=$(( $lngth_lags -1 )); i++))
  do

    result=${lags[$i]}
    echo -e "$timestamp\t$result" >> $log_file
    echo "result piped"

  done
}

for each in "${array[@]}"
do 

   get_lags $each &

done

--------编辑2------------

尝试以下答案:

#!/bin/bash
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/etc/kafka/kafka.client.jaas.conf"
IFS=$'\n'
array=($(kafka-consumer-groups --bootstrap-server kafka1:9092 --list --command-config /etc/kafka/client.properties --new-consumer))

lngth=${#array[*]}

echo "array length: " $lngth

timestamp=$(($(date +%s%N)/1000000))

log_time=`date +%Y-%m-%d:%H`

echo "log time: " $log_time

log_file="/home/ec2-user/laglogs/laglog.$log_time.log"

echo "log file: " $log_file

echo "timestamp: " $timestamp

max_proc_count=8

run_for_each() {
  local each=$1
  echo "Processing: $each" >&2
  IFS=$'\n' read -r -d '' -a lags < <(kafka-consumer-groups --bootstrap-server kafka1:9092 --describe --command-config /etc/kafka/client.properties --new-consumer --group "$each" && printf '\0')
  for result in "${lags[@]}"; do
    printf '%(%Y-%m-%dT%H:%M:%S)T\t%s\t%s\n' -1 "$each" "$result"
  done >>"$log_file"
}

export -f run_for_each
export log_file # make log_file visible to subprocesses

printf '%s\0' "${array[@]}" |
  xargs -P "$max_proc_count" -n 1 -0 bash -c 'run_for_each "$@"'

Answer 1

最方便的做法是将您的后台代码放入一个单独的脚本——或一个导出的函数中。那样xargs可以创建一个新的 shell，并从其父级访问该函数。 (一定要 export 任何其他需要在 child 中可用的变量)。

array=( 1 2 3 4 5 6 )
max_proc_count=8
log_file=out.txt

run_for_each() {
  local each=$1
  echo "Processing: $each" >&2
  IFS=$' \t\n' read -r -d '' -a lags < <(yourcommand --arg1 "$each" && printf '\0')
  for result in "${lags[@]}"; do
    printf '%(%Y-%m-%dT%H:%M:%S)T\t%s\t%s\n' -1 "$each" "$result"
  done >>"$log_file"
}

export -f run_for_each
export log_file # make log_file visible to subprocesses

printf '%s\0' "${array[@]}" |
  xargs -P "$max_proc_count" -n 1 -0 bash -c 'run_for_each "$@"'

一些注意事项:

• 使用 echo -e是错误的形式。请参阅 the POSIX spec for echo 中的应用程序使用和基本原理部分, 明确建议使用 printf相反(不定义 -e 选项，明确定义 echo 不得接受除 -n 以外的任何选项)。
• 我们包括 each日志文件中的值，以便稍后可以从那里提取。
• 您还没有指定 yourcommand 的输出是否是空格分隔、制表符分隔、行分隔或其他方式。因此，我暂时接受所有这些；修改IFS的值传递给了 read品尝。
• printf '%(...)T'无需外部工具即可获取时间戳，例如 date需要 bash 4.2 或更新版本。如果您认为合适，请替换为您自己的代码。
• read -r -a arrayname < <(...)比 arrayname=( $(...) ) 强得多.特别是，它避免将发出的值视为 glob——替换 * s 包含当前目录中的文件列表，或 Foo[Bar]与 FooB是否存在该名称的任何文件(或者，如果设置了 failglob 或 nullglob 选项，则触发失败或在这种情况下根本不发出任何值)。
• 将标准输出重定向到您的 log_file整个循环一次比每次要运行时都重定向它更有效 printf一次。请注意，只有当所有进程都使用 O_APPEND 打开它时，让多个进程同时写入同一个文件才是安全的。 (>> 会这样做)，并且如果他们编写的 block 足够小以作为单个系统调用单独完成(这可能发生，除非单独的 lags 值非常大)。