在Linux系统中使用Shell实现多线程运行任务（多任务并发执行） 2022-05-30

Posted 2023-05-03

参考技术A

最近，有一批任务需要把两批的fastq合并到一起并压缩成一个fastq文件才能继续往下做，由于存储空间有限又不能直接全部跑上，只能按样本逐个分批跑。众所周知，一般fastq是成对存在的，所需要对read1和read2分别合并一次，然而这次任务的fastq文件比较大，合并然后压缩一次需要1天左右，那对于一组fastq就要2-3天，这也太耗时间了，所以我在想能不能read1和read2 同时跑上，这就可以节省一半的时间了。

平时也能遇到很多类似的任务，特别是在进程数有限的情况下，如果这些小任务单独占用一个进程，而任务很多就很耗时间，如果能在一个进程下实现多个线程并行执行，就能大大提高运行效率。关于进程和线程的知识可以参考知乎的这篇文章【 Shell“ 多线程”，提高工作效率 】，整理的也比较有条理，能比较容易读懂。

当然，某些博主也写过类似的文章，例如这篇【 shell后台限制多并发控制后台任务强度进行文件拷贝 】但是实在是太高深莫测了，看不懂，一时半会儿也学不会。本文将示例Shell实现多线程的简单版本，其实不用太复杂。

其实只需要两个步骤， 第一步是给需要并行运行的命令行在结尾加上"&"，代表放到后台运行，第二步是在在所有并行任务的后面加上一句“wait”，意思是等所有通过“&”放到后台运行的任务跑完后再继续执行后面的任务 ，这些就能实现所有带有“&”的行并行执行了。

看完脚本是不是觉得很简单？

上面的脚本适合并行任务少的，可以手动加&和wait，但是如果有几十个甚至上百个的小任务就比较麻烦了。但不用担心，可以写个循环，批量运行。

循环的结果也是跟上面类似的，只是多了个循环结构。

如果需要执行的任务只有一行，可以把大括号去掉。

关于for和while的循环可以查看之前的文章【 Shell常用循环示例（for和while批量处理）2022-05-25 】

需要注意的是多线程并行还是需要有限制的，毕竟都是在一个进程里运行，如果线程太多了会卡顿的，建议控制在100个以内，当然还有毕竟高级和复杂的方法可以实现限制。因为上面的脚本已经够我用了，没继续往下学，以后可以再补充。

Shell 实现多任务并发

实现思路

实现一个shell进程库，通过类似于init，run，wait几个简单的命令，就可以迅速实现多进程并发，伪码如下：

process_init # 创建进程
for city in ${cities[*]}
do
cmd="handler $city"
process_run $cmd 
done
process_wait # 等待进程

原理解析

在实现C++线程库的时候，通常会有一个任务队列，线程从队列中取任务并运行。在实现shell进程库的时候，采用了类似原理，通过一个有名管道充当任务队列。严格来说，并不是一个任务队列，而是一个令牌桶。进程从桶中取得令牌后才可以运行，运行结束后将令牌放回桶中。没有取得令牌的进程不能运行。令牌的数目即允许并发的最大进程数。

 

 

 

 

管道

主要思路：通过mkfifo创建一个有名管道，将管道与一个文件描述符绑定，通过往管道中写数据的方式，控制进程数量。

function _create_pipe()
{
_PROCESS_PIPE_NAME=$(_get_uid)

mkfifo ${_PROCESS_PIPE_NAME}
eval exec "${_PROCESS_PIPE_ID}""<>${_PROCESS_PIPE_NAME}"

for ((i=0; i < $_PROCESS_NUM; i++))
do
echo -ne "\\n" 1>&${_PROCESS_PIPE_ID}
done
}

 

exec

exec fd < file #以读方式打开文件，并关联文件描述符fd
exec fd > file #以写方式打开文件，并关联文件描述符fd
exec fd <> file #以读写方式打开文件，并关联文件描述符

 

# 测试
exec 8>file
echo "hello word!" 1>&8

 

eval
为了让程序有一定的扩展性，不想写死fd，因而引入了变量。

file_fd=8
exec ${file_fd}>file

 

# 测试
bash test.sh
test.sh: line 7: exec: 8: not found

原因：shell在重定向操作符(<、>)左边不进行变量展开。因而引入eval命令，强制shell进行变量展开。 
eval exec "${fd}>file"简单的说，eval将右边参数整体作为一个命令，进行变量的替换，然后将替换后的输出结果给shell去执行。

进程关系

命令执行

function process_run()
{
cmd=$1
if [ -z "$cmd" ]; then
echo "please input command to run"
_delete_pipe 
exit 1
fi

_process_get
{
$cmd
_process_post
}&
}

 

_process_get从管道中取得一个令牌，创建一个进程执行任务，任务执行完毕后，通过_process_post将令牌放回管道。

进程创建

chengsun@153_92:~/test> bash process_util.sh
chengsun@153_92:~/test> pstree -a

|`-sshd
|    |-bash
|    |   `-bash process_util.sh         #爷爷
|    |       |-bash process_util.sh     #爸爸
|    |       |   `-a.out                #儿子
|    |       |-bash process_util.sh        
|    |       |   `-a.out                   
|    |       `-bash process_util.sh
|    |           `-a.out

 

脚本运行后，通过pstree命令，得到如上父子进程关系。稍微做一下解释：当运行脚本bash process_util.sh的时候，创建一个shell进程，相当于爷爷被创建起来，而遇到{ command1; command2 } &时，shell 又创建一个子shell进程（爸爸进程）处理命令序列；而对于每一个外部命令，shell都会创建一个子进程运行该命令，即儿子进程被创建。

困惑：为什么处理{ command1; command2; } &需要单独创建子进程？ 
按照bash manual说法，{ list }并不会创建一个新的shell来运行命令序列。但由于加入&，代表将命令族放入后台执行，就必须新开subshell，否则shell会阻塞。

进程组

 

chengsun@153_92:~/test> ps -f -e -o pid,ppid,pgid,comm

 PID  PPID  PGID  COMMAND
24904 21976 24904 bash
19885 24904 19885  \\_ bash            # 爷爷
19893 19885 19885      \\_ bash        # 爸爸
19894 19893 19885      |   \\_ a.out   # 儿子
19895 19885 19885      \\_ bash
19896 19895 19885      |   \\_ a.out
19897 19885 19885      \\_ bash
19898 19897 19885          \\_ a.out

 

Shell 将运行process_util的一堆进程置于一个进程组中。其中爷爷进程作为该进程组的组长，pid == pgid。

wait
wait pid:阻塞等待某个进程结束； 如果没有指定参数，wait会等待所有子进程结束。

清理函数

允许任务通过CTRL+C方式提前结束，因而需要清理函数

信号

trap 
类似C语言signal函数，为shell脚本注册信号处理函数。trap arg signals，其中signals为注册的信号列表，arg为收到信号后执行某个命令。

 

function Print
{
echo "Hello World!"
}

trap Print SIGKILL

kill 
kill 命令给进程或进程组发送信号；kill pid 给进程发送默认信号SIGTERM, 通知程序终止执行。

 

void sig_handler(int signo)
{
printf("sigterm signal\\n");
}

int main()
{
    signal(SIGTERM, sig_handler);
    sleep(100);

return 0;
}

chengsun@153_92:~/test> ./a.out &
[1] 19254
chengsun@153_92:~/test> kill 19254
sigterm signal

 

kill 0：给当前进程组发送默认信号SIGTERM

chengsun@153_92:~/test> man kill
0  All processes in the current process group are signaled.

清理

 

function _clean_up
{
# 清理管道文件
    _delete_pipe

kill 0
kill -9 $$
}

trap _clean_up SIGINT SIGHUP SIGTERM SIGKILL

 

kill -9 $$ 非常重要

 

 

 

实际上，最上层是爷爷进程，当发送Ctrl + C命令的时候，爷爷进程捕获SIGINT信号，调用_clean_up函数。爷爷进程对整个进程组发送SIGTERM信号，并调用kill -9结束自己。爸爸进程接收SIGTERM信号，同时也发送SIGTERN给整个进程组，如果没有kill -9，爸爸进程始终无法结束，进入无限递归环节。儿子为CPP二进制程序，内部没有捕获SIGTERM，该信号默认动作是结束进程。

使用范例

# file: run.sh
#!/bin/sh

#load process library
source ./process_util.sh

function handler()
{
    city=$1
    ./main ${city}
}

process_init 23
for city in $cities
do
    cmd = "handler $city"
    process_run "$cmd"
done
process_wait

————————————————
版权声明：本文为CSDN博主「spch2008」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/spch2008/article/details/51433353

喜欢这篇文章？欢迎打赏~~