7

使用 bash 和常用工具为 Linux 构建最小任务队列系统的最佳/最简单方法是什么？

我有一个包含 9'000 行的文件，每一行都有一个 bash 命令行，这些命令是完全独立的。

command 1 > Logs/1.log
command 2 > Logs/2.log
command 3 > Logs/3.log
...

我的盒子有多个核心，我想同时执行 X 个任务。我在网上搜索了一个很好的方法来做到这一点。显然，很多人都有这个问题，但到目前为止还没有一个好的解决方案。

如果该解决方案具有以下功能，那就太好了：

  • 可以解释多个命令（例如command; command）
  • 可以解释行上的流重定向（例如ls > /tmp/ls.txt）
  • 只使用常用的 Linux 工具

如果它适用于其他没有太奇特要求的 Unix 克隆，则可以加分。

4

9 回答 9

15

您可以将命令列表转换为 Makefile 吗？如果是这样，您可以运行“make -j X”。

于 2009-05-06T23:23:38.777 回答
11

GNU Parallel http://www.gnu.org/software/parallel/是一个比 PPSS 更通用的并行化工具。

如果运行文件包含：

command 1 > Logs/1.log
command 2 > Logs/2.log
command 3 > Logs/3.log

你可以做：

cat runfile | parallel -j+0

这将为每个 CPU 内核运行一个命令。

如果你的命令和上面一样简单，你甚至不需要运行文件，但可以：

seq 1 3 | parallel -j+0 'command {} > Logs/{}.log'

如果您有更多计算机可用于进行处理，您可能需要查看 GNU Parallel 的 --sshlogin 和 --trc 选项。

于 2010-06-10T20:09:18.417 回答
0

好的，在这里发布问题后，我发现了以下看起来很有希望的项目：ppss。

编辑：不是我想要的，PPSS 专注于处理“目录 A 中的所有文件”。

于 2009-05-06T23:17:48.667 回答
0

好吧，无论如何，这是一个有趣的问题。

这就是我要做的，当然假设是bash(1)。

  • 弄清楚这些命令中有多少可以同时有效地运行。这不仅仅是核心的数量。许多命令将因 I/O 之类的事情而暂停。N=15例如，拨打那个号码 N。
  • 为 SIGCHLD 信号设置一个陷阱信号处理程序，该信号在子进程终止时发生。trap signalHandler SIGCHLD
  • 将您的命令列表放入管道中
  • 编写一个循环，读取标准输入并一一执行命令，递减一个计数器。当计数器为 0 时，它wait是 s。
  • 在 SIGCHLD 上运行的信号处理程序会增加该计数器。

所以现在，它运行第一个N命令，然后等待。当第一个孩子终止时，等待返回，它读取另一行，运行一个新命令，然后再次等待。

现在，这是一个处理许多紧密终止的工作的案例。我怀疑您可以使用更简单的版本：

 N=15
 COUNT=N
 cat mycommands.sh | 
 while read cmd 
 do
   eval $cmd &
   if $((count-- == 0))
   then
       wait
   fi
 od

现在，这个命令将启动前 15 个命令，然后在某个命令终止时一次运行其余命令。

于 2009-05-06T23:45:53.467 回答
0

类似的分布式计算乐趣是 Mapreduce Bash 脚本：

http://blog.last.fm/2009/04/06/mapreduce-bash-script

并感谢您指出 ppss！

于 2009-05-07T08:57:14.233 回答
0

您可以使用xargs命令，它的--max-procs 可以满足您的需求。例如 Charlie Martin 解决方案变成了 xargs：

tr '\012' '\000' <mycommands.sh |xargs --null --max-procs=$X bash -c

细节：

  • X 是最大进程数。例如：X=15。--max-procs 正在发挥作用
  • 第一个 tr 在这里以 xargs --null 选项的空字节终止行，以便引号重定向等不会被错误地扩展
  • bash -c 运行命令

例如，我用这个 mycommands.sh 文件对其进行了测试：

date
date "+%Y-%m-%d" >"The Date".txt
wc -c <'The Date'.txt >'The Count'.txt
于 2009-05-09T08:25:30.650 回答
0

这是一种特殊情况，但是如果您尝试处理一组文件并生成另一组输出文件，则可以启动#cores number of processes，并在处理之前检查输出文件是否存在。下面的示例将 .m4b 文件的目录转换为 .mp3 文件：

只要你有核心运行这个命令的次数：

ls *m4b|同时读取 f; 做测试 -f ${f%m4b}mp3 || mencoder -of rawaudio "$f" -oac mp3lame -ovc 复制 -o ${f%m4b}mp3; 完毕 ＆

于 2009-09-07T08:41:15.420 回答
0

你可以看到我写在 bash 上的任务队列：https ://github.com/pavelpat/yastq

于 2014-11-04T01:08:36.727 回答
0

任务队列+并行化+动态添加

该脚本使用 FIFO 分叉自身来处理队列。这样，您可以动态地将命令添加到队列中（当队列已经启动时）。

用法：./queue 命令 [# of children] [队列名称]

例如，有 1 个线程：

./queue “睡眠 5；回显一”
./queue “回声二”

输出：

一
二

例如，有 2 个线程：

./queue “睡眠 5；回显一” 2
./queue “回声二”

输出：

二
一

例如，有 2 个队列：

./queue "睡眠 5；回显 ONE queue1" 1 queue1
./queue "睡眠 3；回显 ONE queue2" 1 queue2

输出：

一个队列2
一个队列1

脚本（将其保存为“队列”和 chmod +x 队列）：

#!/bin/bash

#Print usage
[[ $# -eq 0 ]] && echo Usage: $0 Command [# of children] [Queue name] && exit

#Param 1 - Command to execute
COMMAND="$1"

#Param 2 - Number of childs in parallel
MAXCHILD=1
[[ $# -gt 1 ]] && MAXCHILD="$2"

#Param 3 - File to be used as FIFO
FIFO="/tmp/defaultqueue"
[[ $# -gt 2 ]] && FIFO="$3"

#Number of seconds to keep the runner active when unused
TIMEOUT=5

runner(){
  #Associate file descriptor 3 to the FIFO
  exec 3<>"$FIFO"

  while read -u 3 -t $TIMEOUT line; do
    #max child check
    while [ `jobs | grep Running | wc -l` -ge "$MAXCHILD" ]; do
      sleep 1
    done

    #exec in backgroud
    (eval "$line")&
  done
  rm $FIFO
}

writer(){
  #fork if the runner is not running
  lsof $FIFO >/dev/null || ($0 "QueueRunner" "$MAXCHILD" "$FIFO" &)

  #send the command to the runner
  echo "$COMMAND" > $FIFO
}

#Create the FIFO file
[[ -e "$FIFO" ]] || mkfifo "$FIFO"

#Start the runner if in the runner fork, else put the command in the queue
[[ "$COMMAND" == "QueueRunner" ]] && runner || writer
于 2015-03-22T20:38:17.807 回答