推薦（免費）：swoole
本來計劃開發 swoft 框架 中的 process 模塊, 所以需要對 swoole 的 process 模塊要有比較深入的了解才行. 不過根據 swoole 官方 wiki 的實踐過程中, 一直有未理解的部分. 之前雖然也做過多次 多進程編程, 但是當真正需要進行框架開發的時候, 就會發現以前學到的知識不夠全面, 無法指導整體的設計. 好在一直在堅持, 奉上現在理解的程度.
內容一覽:
- 進程相關基礎操作: fork/exit/kill/wait
- 進程相關高級操作: 主進程退出子進程干完活后也退出; 子進程異常退出主進程自動重啟
- 進程間通信(IPC) – 管道(pipe)
- 進程間通信(IPC) – 消息隊列(message queue)
- swoole process 模塊提供的更多功能
進程相關基礎操作
進程是什么: 進程是運行者的程序
先來看看一個最簡單的例子:
<?phpecho posix_getpid(); // 獲取當前進程的 pidswoole_set_process_name('swoole process master'); // 修改所在進程的進程名sleep(100); // 模擬一個持續運行 100s 的程序, 這樣就可以在進程中查看到它, 而不是運行完了就結束
通過 ps aux 查看進程:
未設置進程名
設置進程名
再來看看 swoole 中使用子進程的基礎操作:
use?SwooleProcess; $process?=?new?Process(function?(Process?$worker)?{????if?(Process::kill($worker->pid,?0))?{?//?kill操作常用來殺死進程,?傳入?0?可以用來檢測進程是否存在 ????????$worker->exit();?//?退出子進程 ????} }); $process->start();?//?啟動子進程Process::wait();?//?回收退出的子進程
-
new Process(): 通過回調函數來設置子進程將要執行的邏輯
-
$process->start(): 調用 fork() 系統調用, 來生成子進程
-
Process::kill(): kill操作給進程發送信號, 常用來殺死進程, 傳入 0 可以用來檢測進程是否存在
-
Process::wait(): 調用 wait() 系統調用, 回收子進程, 如果不回收, 子進程會編程 僵尸進程, 浪費系統資源
-
$worker->exit(): 子進程主動退出
我在這里有一個疑問:
主進程的生命周期是怎么樣的? 子進程的生命周期是怎么樣的?
有這樣一個疑問也來自于我之前的思維慣性: 理解一個事物時從事物的生命周期進行理解. 結合 進程是運行著的程序 來一起理解:
- new Process(): 只有回調函數的邏輯會在進程中執行
- 除此之外的代碼都是在主進程中執行
進程相關高級操作
- 主進程退出子進程干完活后也退出
- 子進程異常退出主進程自動重啟
<?phpuse SwooleProcess;class MyProcess1{ public $mpid = 0; // master pid, 即當前程序的進程ID public $works = []; // 記錄子進程的 pid public $maxProcessNum = 1; public $newIndex = 0; public function __construct() { try { swoole_set_process_name(__CLASS__. ' : master'); $this->mpid?=?posix_getpid();????????????$this->run();????????????$this->processWait(); ????????}?catch?(Exception?$e)?{????????????die('Error:?'.?$e->getMessage()); ????????} ????}????public?function?run() ????{????????for?($i=0;?$imaxProcessNum;?$i++)?{????????????$this->createProcess(); ????????} ????}????public?function?createProcess($index?=?null) ????{????????if?(is_null($index))?{ ????????????$index?=?$this->newIndex;????????????$this->newIndex++; ????????} ????????$process?=?new?Process(function?(Process?$worker)?use($index)?{?//?子進程創建后需要執行的函數 ????????????swoole_set_process_name(__CLASS__.?":?worker?$index");????????????for?($j=0;?$jcheckMpid($worker);????????????????echo?"msg:?{$j}n"; ????????????????sleep(1); ????????????} ????????},?false,?false);?//?不重定向輸入輸出;?不使用管道 ????????$pid?=?$process->start();????????$this->works[$index]?=?$pid;????????return?$pid; ????}????//?主進程異常退出,?子進程工作完后退出 ????public?function?checkMpid(Process?$worker)?//?demo中使用的引用,?引用表示傳的參數可以被改變,?由于傳入?$worker?是?SwooleProcess?對象,?所以不用使用?&????{????????if?(!Process::kill($this->mpid,?0))?{?//?0?可以用來檢測進程是否存在 ????????????$worker->exit(); ????????????$msg?=?"master?process?exited,?worker?{$worker->pid}?also?quitn";?//?需要寫入到日志中 ????????????file_put_contents('process.log',?$msg,?FILE_APPEND);?//?todo:?這句話沒有執行 ????????} ????}????//?重啟子進程 ????public?function?rebootProcess($pid) ????{ ????????$index?=?Array_search($pid,?$this->works);????????if?($index?!==?false)?{ ????????????$newPid?=?$this->createProcess($index);????????????echo?"rebootProcess:?{$index}={$pid}->{$newPid}?Donen";????????????return; ????????}????????throw?new?Exception("rebootProcess?error:?no?pid?{$pid}"); ????}????//?自動重啟子進程 ????public?function?processWait() ????{????????while?(1)?{????????????if?(count($this->works))?{ ????????????????$ret?=?Process::wait();?//?子進程退出 ????????????????if?($ret)?{????????????????????$this->rebootProcess($ret['pid']); ????????????????} ????????????}?else?{????????????????break; ????????????} ????????} ????} }new?MyProcess1();
說明以下幾點:
- 子進程運行結束后就會退出, 通過 Process::wait() 檢測到子進程退出信號執行自動重啟, 子進程就會一直執行下去
- 關于函數參數傳 引用/指針, 一個很好的理解方式是: 參數可以被修改
運行并模擬主進程異常退出:
模擬主進程異常退出
輸出
進程間通信(IPC) – 管道(pipe)
管道的幾個關鍵詞:
- 半雙工: 數據單向流動, 一端只讀, 一端只寫.
- 同步 vs 異步: 默認為同步阻塞模式, 可以使用 swoole_Event_add() 添加管道到 swoole 的 event loop 中, 實現異步IO
- 管道類型(數據格式): SOCK_STREAM, 流式, 需要用戶自己處理數據的封包/解包; SOCK_DGRAM, 數據報, 每次收發都是一次完整的數據包 (DGRAM/STREAM)
注意, swoole wiki – process->write() 中提到 SOCK_DGRAM 并不會亂序丟包
先來看一個簡單的例子, php從shell管道中讀取數據:
//?get?pip?data$fp?=?fopen('php://stdin',?'r');if?($fp)?{????while?($line?=?fgets($fp,?4096))?{????????echo?"php?get?pip?data:?".?$line; ????} ????fclose($fp); }
從shell管道讀取數據
swoole process中的管道很強大, 支持 子進程寫, 主進程讀 以及 主進程寫, 子進程讀:
use?SwooleProcess;//?子進程寫,?父進程讀$process?=?new?Process(function?(Process?$worker)?{ ????$worker->write("worker"); }); $process->start(); $msg?=?$process->read();echo?"from?process:?$msg",?"n";//?父進程寫,?子進程讀$process?=?new?Process(function?(Process?$worker)?{ ????$msg?=?$worker->read();????echo?"from?master:?$msg",?"n"; }); $process->start(); $process->write('master');
使用管道多次讀寫
注意區分 $worker->write() 和 $process->write(), 之前一直錯誤的以為這 2 個是相同的, 其實就是把 $process 誤以為是子進程, 從而相當于 $process->write() 就是子進程寫管道 — 其實這里是主進程內執行的邏輯, 是主進程寫數據到管道, 供子進程讀取
swoole中其他管道相關操作:
- 異步IO
use?SwooleProcess;use?SwooleEvent;//?異步IO$process?=?new?Process(function?(Process?$worker)?{ ????$GLOBALS['worker']?=?$worker; ????Event::add($worker->pipe,?function?(int?$pipe)?{?//?使用?swoole_event_add?添加管道到異步IO ????????/**?@var?Process?$worker?*/ ????????$worker?=?$GLOBALS['worker']; ????????$msg?=?$worker->read();????????echo?"from?master:?$msg?n"; ????????$worker->write("hello?master"); ????????sleep(2); ????????$worker->exit(0); ????}); }); $process->start(); $process->write("master?msg?1"); $msg?=?$process->read();echo?"from?process:?$msg?n";
異步IO
- 設置超時
use?SwooleProcess;//?設置管道超時$process?=?new?Process(function?(Process?$worker)?{ ????sleep(5); }); $process->start(); $process->setTimeout(0.5); $ret?=?$process->read(); var_dump($ret); var_dump(swoole_errno());
管道超時
插播一個趣事, @thinkpc 看完 2017北京PHP開發者年會, 就知道為啥會點贊了
- 關閉管道
// 關閉管道: 默認值0->關閉讀寫 1->關閉寫 2->關閉讀$process->close();
進程間通信(IPC) – 消息隊列(message queue)
消息隊列:
- 一系列保存在內核中的消息鏈表
- 有一個 msgKey, 可以通過此訪問不同的消息隊列
- 有數據大小限制, 默認 8192, 可以通過內核修改
- 阻塞 vs 非阻塞: 阻塞模式下 pop()空消息隊列/push()滿消息隊列會阻塞, 非阻塞模式可以直接返回
swoole 中使用消息隊列:
- 通信模式: 默認為爭搶模式, 無法將消息投遞給指定子進程
- 新建消息隊列后, 主進程就可以使用
- 消息隊列不可和管道一起使用, 也無法使用 swoole event loop
- 主進程中要調用 wait(), 否則子進程中調用 pop()/push() 會報錯
use?SwooleProcess; $process?=?new?Process(function?(Process?$worker)?{????//?$worker->push('worker'); ????echo?"from?master:?".?$worker->pop().?"n"; ????sleep(2);????//?$worker->exit();},?false,?false);?//?關閉管道//?參數一為?msgKey,?這里是默認值//?參數二為?通信模式,?默認值?2?表示爭搶模式,?這里還加上了?非阻塞$process->useQueue(ftok(__FILE__,?1),?2|?Process::IPC_NOWAIT); $process->push('hello1');?//?使用?useQueue?后,?主進程就可以讀寫消息隊列了$process->push('hello2');echo?"from?woker:?".?$process->pop().?"n";//?echo?"from?woker:?".?$process->pop().?"n";$process->start();?//?啟動子進程//?消息隊列狀態var_dump($process->statQueue());//?刪除隊列,?如果不調用則不會在程序結束時清楚數據,?下次使用相同?msgKey?時還可以訪問數據$process->freeQueue(); var_dump(Process::wait());?//?要調用?wait(),?否則子進程中?push()/pop()?會報錯
消息隊列
swoole process 模塊提供的更多功能
-
swoole_set_process_name(): 修改進程名, 不兼容 mac
-
swoole_process->exec(String $execfile, array $args) 執行外部程序
參數 $execfile 需要使用可執行文件的絕對路徑, 參數 args 為參數數組
//?比如?python?test.py?123swoole_process->exec('/usr/bin/python',?['test.py',?123]);//?更復雜的例子swoole_process->exec(('/usr/local/bin/php',?['/var/www/project/yii-best-practice/cli/yii',?'t/index',?'-m=123',?'abc',?'xyz']);//?父進程?exec?進程進行管道通信use?SwooleProcess; $process?=?new?Process(function?(Process?$worker)?{ ????$worker->exec('/bin/echo',?['hello']); ????$worker->write('hello'); },?true);?//?需要啟用標準輸入輸出重定向$process->start();echo?"from?exec:?".?$process->read().?"n";
父進程與exec進程通過管道通信
-
SwooleProcess::kill($pid, $signo = SIGTERM): 向指定進程發送信號, 默認是終止進程, 傳 0 可檢測進程是否存在
-
SwooleProcess::wait(): 回收子進程, 如果主進程不調用此方法, 子進程會變成 僵尸進程, 浪費系統資源
-
SwooleProcess::signal(): 異步信號監聽
use?SwooleProcess;//?異步信號監聽?+?waitProcess::signal(SIGCHLD,?function?($signal)?{?//?監聽子進程退出信號 ????//?可能同時有多個子進程退出,?所以要while循環 ????while?($ret?=?Process::wait(false))?{?//?false?表示不阻塞 ????????var_dump($ret); ????} });
SwooleProcess::daemon(): 將當前進程變為一個守護進程
use?SwooleProcess;//?daemonProcess::daemon(); swoole_set_process_name('test?daemon?process'); sleep(100);
daemon-守護進程
- SwooleProcess::alarm(): 高精度定時器(微秒級), 對 setitimer 系統調用的封裝, 可以配合 SwooleProcess::signal() / pcntl_signal 使用
注意不可和 SwooleTimer 同時使用
//?signal?+?alarm//?第一個參數表示時間,?單位?us,?-1?表示清除定時器//?第二個參數表示類型?0->真實時間->SIGALAM?1->cpu時間->SIGVTALAM?2->用戶態+內核態時間->SIGPROFProcess::alarm(100*1000);?//?100msProcess::signal(SIGALRM,?function?($signal)?{????static?$i?=?0;????echo?"#$i?t?alarm?n"; ????$i++;????if?($i>20)?{ ????????Process::alarm(-1);?//?-1?表示清除 ????} });
alarm
- SwooleProcess::setaffinity(): 設置CPU親和, 即將進程綁定到指定CPU核上
傳值范圍: [0, swoole_cpu_num())
CPU親和: CPU的速度遠遠高于IO的速度, 所以CPU有多級緩存來解決IO等待的問題, 綁定指定CPU, 更容易命中CPU緩存
寫在最后
資源推薦:
- 圖靈社區 – 理解unix進程 + 「理解Unix進程」讀書筆記
- blog – 「進程」編程
todo:
- 使用輸入輸出重定向
- 管道類型為 SOCK_STREAM 時的情況, 是否需要 封包/解包 處理, 即 swoole wiki – process->write() 中提到的 管道通信默認的方式是流式，write寫入的數據在read可能會被底層合并
- 多進程 + 異步IO 的注意事項
能理解 因為子進程會繼承父進程的內存和IO句柄 這個會產生的影響, 但是給的示例并沒有說明這個問題
use?SwooleProcess;use?SwooleEvent;//?多個子進程?+?異步IO$workers?=?[]; $workerNum?=?3;for?($i=0;?$iwrite($worker->pid);????????echo?"worker:?{$worker->pid}?n"; ????}); ????$pid?=?$process->start(); ????$workers[$pid]?=?$process;????//?Event::add($process->pipe,?function?(int?$pipe)?use?($process)?{ ????//??$data?=?$process->read(); ????//??echo?"recv:?$data?n"; ????//?});}foreach?($workers?as?$worker)?{ ????Event::add($worker->pipe,?function?(int?$pipe)?use?($worker)?{ ????????$data?=?$worker->read();????????echo?"recv:?$data?n"; ????}); }
多進程異步IO
? 版權聲明
文章版權歸作者所有，未經允許請勿轉載。
THE END
