摘要：本文使用與完成一個(gè)的進(jìn)程池，并且支持動(dòng)態(tài)創(chuàng)建新進(jìn)程。接著遍歷所有的進(jìn)程，并且加入中，設(shè)置可讀事件，用于接收子進(jìn)程的空閑信號(hào)。最后每隔一秒向進(jìn)程投遞任務(wù)。由于只模擬了十次任務(wù)，則第十個(gè)任務(wù)完成之后在父進(jìn)程中發(fā)送使所有子進(jìn)程退出。

swoole_process 主要是用來代替 PHP 的 pcntl 擴(kuò)展。我們知道 pcntl 是用來進(jìn)行多進(jìn)程編程的，而 pcntl 只提供了 fork 這樣原始的接口，容易使用錯(cuò)誤，并且沒有提供進(jìn)程間通信以及重定向標(biāo)準(zhǔn)輸入輸出的功能。

而 swoole_process 則提供了比 pcntl 更強(qiáng)大的功能，更易用的API，使PHP在多進(jìn)程編程方面更加輕松。

本文使用 swoole_process 與 EventLoop 完成一個(gè) php 的進(jìn)程池，并且支持動(dòng)態(tài)創(chuàng)建新進(jìn)程。

swoole 有一個(gè) Reactor 線程，這個(gè)線程可以說是對(duì) epoll 模型的封裝，可以設(shè)置 read 事件和 write 事件的監(jiān)聽回調(diào)函數(shù)。

下面會(huì)用到一個(gè)函數(shù)：

bool swoole_event_add(mixed $sock, mixed $read_callback, mixed $write_callback = null, int $flags = null);

參數(shù)1為一個(gè)文件描述符，包括swoole_client->$sock、swoole_process->$pipe或者其他 fd（socket_create 創(chuàng)建的資源 ， stream_socket_client/fsockopen創(chuàng)建的資源）

參數(shù)2為可讀事件回調(diào)函數(shù)

參數(shù)3為可寫事件回調(diào)函數(shù)

多進(jìn)程編程少不了進(jìn)程之間的通訊，swoole 的進(jìn)程之間有兩種通信方式，一種是消息隊(duì)列(queue)，另一種是管道(pipe)。那么本文使用的是 pipe 的方式。

下面是一個(gè)定時(shí)向進(jìn)程池投遞任務(wù)的例子。

代碼：

process = new swoole_process(array($this, "run"), false, 2);
        $this->process->start();
        swoole_process::wait();
    }

    public function run()
    {
        $this->current_num = $this->min_worker_num;
        //創(chuàng)建所有的worker進(jìn)程
        for($i = 0; $i < $this->current_num; $i++){
            $process = new swoole_process(array($this, "task_run"), false, 2);
            $pid = $process->start();
            $this->process_list[$pid] = $process;
            $this->process_use[$pid] = 0;
        }

        foreach($this->process_list as $process){
            swoole_event_add($process->pipe, function ($pipe) use ($process){
                $data = $process->read();
                var_dump($data . "空閑");
                //接收子進(jìn)程處理完成的信息，并且重置為空閑
                $this->process_use[$data] = 0;
            });
        }

        //每秒定時(shí)向worker管道投遞任務(wù)
        swoole_timer_tick(1000 ,function ($timer_id){
            static $index = 0;
            $index = $index + 1;
            $flag = true; //是否新建worker
            foreach ($this->process_use as $pid => $used){
                if($used == 0){
                    $flag = false;
                    //標(biāo)記為正在使用
                    $this->process_use[$pid] = 1;
                    // 在父進(jìn)程內(nèi)調(diào)用write，子進(jìn)程可以調(diào)用read接收此數(shù)據(jù)
                    $this->process_list[$pid]->write($index. "hello");
                    break;
                }
            }

            if($flag && $this->current_num < $this->max_worker_num){
                //沒有閑置worker，新建worker來處理
                $process = new swoole_process(array($this, "task_run"), false, 2);
                $pid = $process->start();
                $this->process_list[$pid] = $process;
                $this->process_use[$pid] = 1;
                $this->process_list[$pid]->write($index. "hello");
                $this->current_num++;
            }
            var_dump("第" .$index. "個(gè)任務(wù)");
            if($index == 10){
                foreach($this->process_list as $process){
                    $process->write("exit");
                }
                swoole_timer_clear($timer_id);
                $this->process->exit();
            }

        });
    }

    /**
     * 子進(jìn)程處理
     * @param $worker
     */
    public function task_run($worker)
    {
        swoole_event_add($worker->pipe, function($pipe)use($worker){
            $data = $worker->read();
            var_dump($worker->pid . ":" . $data);
            if($data == "exit"){
                $worker->exit();
                exit;
            }
            //模擬耗時(shí)任務(wù)
            sleep(5);
            //告訴主進(jìn)程處理完成
            //在子進(jìn)程內(nèi)調(diào)用write，父進(jìn)程可以調(diào)用read接收此數(shù)據(jù)
            $worker->write($worker->pid);
        });
    }

}

new ProcessPool();

首先定義幾個(gè)重要的屬性：

$process_list ：Worker 進(jìn)程數(shù)組

$process_use：正在被使用的進(jìn)程

$min_worker_num ：最少進(jìn)程數(shù)量

$max_worker_num ：最多進(jìn)程數(shù)量

$current_num ：當(dāng)前進(jìn)程數(shù)量

$process ： 主進(jìn)程

在實(shí)例化的時(shí)候創(chuàng)建主進(jìn)程，并且運(yùn)行 run 方法，在 run 方法里面先創(chuàng)建所有的 worker 進(jìn)程，并且設(shè)置為空閑狀態(tài)。

接著遍歷所有的 worker 進(jìn)程，并且加入 EventLoop 中，設(shè)置可讀事件，用于接收子進(jìn)程的空閑信號(hào)。

最后每隔一秒向 worker 進(jìn)程投遞任務(wù)。動(dòng)態(tài)擴(kuò)充進(jìn)程池則在這里實(shí)現(xiàn)，如果沒有閑置的進(jìn)程，而此時(shí)又有新的任務(wù)，則需要?jiǎng)討B(tài)創(chuàng)建一個(gè)新的進(jìn)程并且置為繁忙狀態(tài)。由于只模擬了十次任務(wù)，則第十個(gè)任務(wù)完成之后在父進(jìn)程中發(fā)送 exit 使所有子進(jìn)程退出。

運(yùn)行效果與圖解：

參考鏈接：

https://wiki.swoole.com/wiki/...
https://opso.coding.me/2018/0...