Linux-Shell-使用mkfifo實現多任務並發及並發數控制

默認的情況下，Shell腳本中的命令是串行執行的，必須等到前一條命令執行完后才執行接下來的命令，但是如果我有一大批的的命令需要執行，而且互相又沒有影響的情況下（有影響的話就比較復雜了），那么就要使用命令的並發執行了。

如下：

#!/bin/bash
IPLIST=/home/meta/ipinfo/iplist
for i in $(cat ${IPLIST} |grep -viE "^#|備機|ts"|awk '{print $1}')
do
ssh $i "cd ~/update/;tar zxf patch-20160909.tgz -C ~/LMDG/ && echo '/$i ok' || echo '/$i bad'"
done >> result.txt

echo "resutl"|mutt -a result.txt -s update-result [e-mail]

 

對於上面的代碼，因為 iplist 中有好多ip，每個”tar zxf”都挺耗時的，所以打算使用並發編程，這樣就可以節省大量時間了。

修改如下：

#!/bin/bash
IPLIST=/home/meta/ipinfo/iplist
for i in $(cat ${IPLIST} |grep -viE "^#|備機|ts"|awk '{print $1}')
do
ssh $i "cd ~/update/;tar zxf patch-20160909.tgz -C ~/LMDG/ && echo '/$i ok' || echo '/$i bad'" &
done >> result.txt

echo "resutl"|mutt -a result.txt -s update-result [e-mail]

 

加上“&” 之后 “tar zxf”就可以並行執行了。 實質是將”tar zxf” 作為后台進程在執行，這樣該命令就不會占用當前bash，其他命令也不用等待前面命令執行完再繼續了，而且可以放入多個任務到后台，這樣就實現了多任務並發。

我本來目的是讓”tar zxf”這個循環都執行結束后，再“mutt”前面的結果。如果像上面這樣寫的話，在”tar zxf”都還沒結束時就已經開始執行“mutt”了，得到了錯誤的結果，因此需要做如下修改：

#!/bin/bash
IPLIST=/home/meta/ipinfo/iplist
for i in $(cat ${IPLIST} |grep -viE "^#|備機|ts"|awk '{print $1}')
do
ssh $i "cd ~/update/;tar zxf patch-20160909.tgz -C ~/LMDG/ && echo '/$i ok' || echo '/$i bad'" &
done >> result.txt
wait
echo "resutl"|mutt -a result.txt -s update-result [e-mail]

 

這里添加了“wait” 之后就可以達到我們預期的效果了，wait的作用就是等待子任務都執行完之后在結束父任務，繼而執行下面的任務。

但是，緊接着又有問題了，如果這個iplist中的量巨大，這樣一口氣都放到后台，系統超出負載后，會有性能變差或者宕機風險，因此我們需要一個控制並發數的機制。
因此我們引入了任務隊列的概念，有點類似之前socket舉例中的消費者生產者模型，通過消息隊列來調節供需的不平衡
修改如下：

#!/bin/bash

IPLIST=/home/meta/ipinfo/iplist #任務（消費者）
THREAD=50 #聲明並發線程並發個數，這個是此應用的關鍵，也就是設置管道的最大任務數
TMPFIFO=/tmp/$$.fifo    #聲明管道名稱，'$$'表示腳本當前運行的進程PID
mkfifo $TMPFIFO #創建管道
exec 5<>${TMPFIFO} #創建文件標示符“5”，這個數字可以為除“0”、“1”、“2”之外的所有未聲明過的字符，以讀寫模式操作管道文件；系統調用exec是以新的進程去代替原來的進程，但進程的PID保持不變，換句話說就是在調用進程內部執行一個可執行文件
rm -rf ${TMPFIFO} #清除創建的管道文件

#為並發線程創建同樣個數的占位
for((i=1;i<=$THREAD;i++))
do
echo ; 
#借用read命令一次讀取一行的特性，使用一個echo默認輸出一個換行符，來確保每一行只有一個線程占位；這里讓人聯想到生產者&消費者模型，管道文件充當消息隊列，來記錄消費者的需求，然后由生產者去領任務，並完成任務，這里運用了異步解耦的思想。
done >&5 
#將占位信息寫入管道

for i in $(cat ${IPLIST} |grep -viE "^#|備機|ts"|awk '{print $1}') #從任務隊列中依次讀取任務
do
read -u5 
#從文件描述符管道中，獲取一個管道的線程占位然后開始執行操作；read中 -u 后面跟fd，表示從文件描述符中讀入，該文件描述符可以是exec新開啟的。
{
echo $(cat ~/ipinfo/iplist|grep $i|awk '{print $2}');
ssh -oConnectTimeout=10 -oConnectionAttempts=3 $i "cd /home/Log/;grep 'MIL' mission_2016-08-03*.log |awk -F, '{if(\$19==1370) print \$0}'|
awk -F, '{if(\$20==0) print \$0}'>miss_info.txt"
echo "" >&5 
#任務執行完后在fd5中寫入一個占位符，以保證這個線程執行完后，線程繼續保持占位，繼而維持管道中永遠是50個線程數，&表示該部分命令/任務放入后台不占當前的bash，實現並行處理
} &
done
wait #等待父進程的子進程都執行結束后再結束父進程 
exec 5>&- #關閉fd5的管道
exit 0

 

————————————————
版權聲明：本文為CSDN博主「白金牧場」的原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/qq_34409701/article/details/52488964

喜歡這篇文章？歡迎打賞~~