swoole启动流程_Swoole+Redis实现并发队列处理

/*** 启动守护进程*/
public function runAction() {Tools::log_message(&＃39;ERROR&＃39;, &＃39;daemon/run&＃39; . &＃39; | action: restart&＃39;, &＃39;daemon-&＃39;);while (true) {$this->fork_process();}exit;
}/*** 创建子进程*/
private function fork_process() {$ppid &＃61; getmypid();$pid &＃61; pcntl_fork();if ($pid &＃61;&＃61; 0) {//子进程$pid &＃61; posix_getpid();//echo "* Process {$pid} was created nn";$this->mq_process();exit;} else {//主进程$pid &＃61; pcntl_wait($status, WUNTRACED); //取得子进程结束状态if (pcntl_wifexited($status)) {//echo "nn* Sub process: {$pid} exited with {$status}";//Tools::log_message(&＃39;INFO&＃39;, &＃39;daemon/run succ&＃39; . &＃39;|status:&＃39; . $status . &＃39;|pid:&＃39; . $ppid . &＃39;|childpid:&＃39; . $pid );} else {Tools::log_message(&＃39;ERROR&＃39;, &＃39;daemon/run fail&＃39; . &＃39;|status:&＃39; . $status . &＃39;|pid:&＃39; . $ppid . &＃39;|childpid:&＃39; . $pid, &＃39;daemon-&＃39;);}}
}/*** 业务任务队列处理*/
private function mq_process() {$data_pop &＃61; $this->masterRedis->rPop($this->redis_list_key);$data &＃61; json_decode($data_pop, 1);if (!$data) {return FALSE;}$worker &＃61; &＃39;_task_&＃39; . $data[&＃39;worker&＃39;];$class_name &＃61; isset($data[&＃39;class&＃39;]) ? $data[&＃39;class&＃39;] : &＃39;TaskproModel&＃39;;$params &＃61; $data[&＃39;params&＃39;];$class &＃61; new $class_name();$class->$worker($params);return TRUE;
}


// install signal handler for dead kids
pcntl_signal(SIGCHLD, array($this, "sig_handler"));


declare(ticks &＃61; 1);


declare(ticks &＃61; 1);
class JobDaemonController extends Yaf_Controller_Abstract{use Trait_Redis;private $maxProcesses &＃61; 800;private $child;private $masterRedis;private $redis_task_wing &＃61; &＃39;task:wing&＃39;; //待处理队列public function init(){// install signal handler for dead kidspcntl_signal(SIGCHLD, array($this, "sig_handler"));set_time_limit(0);ini_set(&＃39;default_socket_timeout&＃39;, -1); //队列处理不超时,解决redis报错:read error on connection}private function redis_client(){$rds &＃61; new Redis();$rds->connect(&＃39;redis.master.host&＃39;,6379);return $rds;}public function process(swoole_process $worker){// 第一个处理$GLOBALS[&＃39;worker&＃39;] &＃61; $worker;swoole_event_add($worker->pipe, function($pipe) {$worker &＃61; $GLOBALS[&＃39;worker&＃39;];$recv &＃61; $worker->read(); //send data to mastersleep(rand(1, 3));echo "From Master: $recvn";$worker->exit(0);});exit;}public function testAction(){for ($i &＃61; 0; $i <10000; $i&＃43;&＃43;){$data &＃61; [&＃39;abc&＃39; &＃61;> $i,&＃39;timestamp&＃39; &＃61;> time().rand(100,999)];$this->masterRedis->lpush($this->redis_task_wing, json_encode($data));}exit;}public function runAction(){while (1){
// echo "t now we de have $this->child child processesn";if ($this->child <$this->maxProcesses){$rds &＃61; $this->redis_client();$data_pop &＃61; $rds->brpop($this->redis_task_wing, 3);//无任务时,阻塞等待if (!$data_pop){continue;}echo "t Starting new child | now we de have $this->child child processesn";$this->child&＃43;&＃43;;$process &＃61; new swoole_process([$this, &＃39;process&＃39;]);$process->write(json_encode($data_pop));$pid &＃61; $process->start();}}}private function sig_handler($signo) {
// echo "Recive: $signo rn";switch ($signo) {case SIGCHLD:while($ret &＃61; swoole_process::wait(false)) {
// echo "PID&＃61;{$ret[&＃39;pid&＃39;]}n";$this->child--;}}}
}