/**
  * Start consumers workers
  */
 public function startConsumersAction()
 {
     $workers = array('feedWorkers' => array('queue' => Queue::QUEUE_FEED_CRAWL, 'callback' => 'crawlFeed'), 'postWorkers' => array('queue' => Queue::QUEUE_POST_PROCEED, 'callback' => 'proceedPost'), 'commentWorkers' => array('queue' => Queue::QUEUE_COMMENTS_CRAWL, 'callback' => 'crawlComments'), 'imageWorkers' => array('queue' => Queue::QUEUE_IMAGE_CRAWL, 'callback' => 'crawlImage'));
     $consumerProcess = new \Importer\ConsumerProcess($this->plugin);
     $importer = new \Importer\Service($this->plugin);
     foreach ($workers as $workerName => $worker) {
         $workersNumber = $this->plugin->config->importer->{$workerName};
         for ($i = 0; $i < $workersNumber; $i++) {
             $pid = $consumerProcess->addConsumer($worker['queue'], array($importer, $worker['callback']));
             if ($pid) {
                 $childs[$pid] = $workerName;
             }
         }
     }
     while (pcntl_waitpid(0, $status) != -1) {
         echo '++';
         foreach ($childs as $pid => $role) {
             $childStatus = pcntl_waitpid($pid, $status, WNOHANG);
             //child died
             if ($childStatus == -1 || $childStatus > 0) {
                 $workerName = $childs[$pid];
                 $worker = $workers[$workerName];
                 unset($childs[$pid]);
                 $pid = $consumerProcess->addConsumer($worker['queue'], array($importer, $worker['callback']));
                 $childs[$pid] = $worker;
                 echo 'restarted worker ', $worker, "\n";
             }
         }
     }
 }

 /**
  * start crontab and loop
  */
 public function start()
 {
     $this->logger->info("crontab start");
     $crontab = $this->createCrontab();
     $loop = Factory::create();
     // add periodic timer
     $loop->addPeriodicTimer(1, function () use($crontab) {
         $pid = pcntl_fork();
         if ($pid > 0) {
             return;
         } elseif ($pid == 0) {
             $crontab->start(time());
             exit;
         } else {
             $this->logger->error("could not fork");
             exit;
         }
     });
     // recover the sub processes
     $loop->addPeriodicTimer(60, function () {
         while (($pid = pcntl_waitpid(0, $status, WNOHANG)) > 0) {
             $message = "process exit. pid:" . $pid . ". exit code:" . $status;
             $this->logger->info($message);
         }
     });
     $loop->run();
 }

 /**
  * Run a callback in $threadsCount parallel processes
  *
  * @param callable $callback Callback to run
  * @param array[] $argsCollection Array of arguments, two-dimensional
  * @return array (pid => callback result)
  */
 public function run(callable $callback, array $argsCollection)
 {
     $file = tempnam(sys_get_temp_dir(), 'php');
     file_put_contents($file, "<?php\n");
     $children = [];
     foreach ($argsCollection as $key => $args) {
         $pid = pcntl_fork();
         switch ($pid) {
             case -1:
                 throw new RuntimeException(sprintf('Unable to fork thread %d of %d', $key, count($argsCollection)));
             case 0:
                 // child
                 $this->runChild($callback, $args, $file);
                 die(0);
             default:
                 //parent
                 $children[] = $pid;
         }
     }
     foreach ($children as $child) {
         pcntl_waitpid($child, $status);
     }
     $result = [];
     require $file;
     unlink($file);
     return $result;
 }

 public static function run()
 {
     if (!function_exists('pcntl_fork')) {
         throw new \Exception('pcntl ext not exists');
     }
     $batchArray = self::getBatchArray();
     $pids = array();
     for ($i = 0; $i < self::$forkCount; $i++) {
         $pids[$i] = pcntl_fork();
         if ($pids[$i] == -1) {
             throw new \Exception('analyzer couldn`t fork');
         } elseif (!$pids[$i]) {
             if (is_array($batchArray[$i]) && count($batchArray[$i]) > 0) {
                 foreach ($batchArray[$i] as $rName => $config) {
                     self::analyzerRun($rName, $config);
                 }
             }
             exit(0);
         }
         if (self::$waitFlag) {
             pcntl_waitpid($pids[$i], $status, WUNTRACED);
             echo "wait {$i} -> " . time() . "\n";
         }
     }
 }

function excute($url = '', $data)
{
    connectRedis();
    $response = array();
    $loop = count($data);
    for ($i = 0; $i < $loop; $i++) {
        # code...
        $pid = pcntl_fork();
        if (!$pid) {
            // sleep(1);
            print "In child {$i}\n";
            if (array_key_exists($i, $data)) {
                $x = 7;
                $k = get_url($data[$i]);
                setRedis("data", $i);
            }
            exit($i);
        }
    }
    #process
    while (pcntl_waitpid(0, $status) != -1) {
        $status = pcntl_wexitstatus($status);
        echo "Child {$status} completed\n";
    }
    // dd($response);
    return $response;
}

 /**
  * recover the sub processes
  */
 public function processRecoverCallback()
 {
     while (($pid = pcntl_waitpid(0, $status, WNOHANG)) > 0) {
         $message = "process exit. pid:" . $pid . ". exit code:" . $status;
         $this->logger->info($message);
     }
 }

function pleac_Gathering_Output_from_a_Program()
{
    // Run a command and return its results as a string.
    $output_string = shell_exec('program args');
    // Same as above, using backtick operator.
    $output_string = `program args`;
    // Run a command and return its results as a list of strings,
    // one per line.
    $output_lines = array();
    exec('program args', $output_lines);
    // -----------------------------
    // The only way to execute a program without using the shell is to
    // use pcntl_exec(). However, there is no way to do redirection, so
    // you can't capture its output.
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('cannot fork');
    } elseif ($pid) {
        pcntl_waitpid($pid, $status);
    } else {
        // Note that pcntl_exec() automatically prepends the program name
        // to the array of arguments; the program name cannot be spoofed.
        pcntl_exec($program, array($arg1, $arg2));
    }
}

 /**
  * Wait for all child processes to complete
  */
 public function wait()
 {
     // Wait for all children to return
     foreach ($this->child_pid_list as $child_pid) {
         pcntl_waitpid($child_pid, $status);
     }
 }

 public function testRunningDaemonWithResistingWorker()
 {
     $writer = $this->getFileWriter();
     $fork = $this->outerManager->fork(function () use($writer) {
         $handler = new NullHandler();
         $builder = new Builder(array('worker1' => function () use($writer) {
             $writer("worker1.call");
         }, 'worker2' => array('startup' => function () use($writer) {
             pcntl_signal(SIGQUIT, SIG_IGN);
             pcntl_signal(SIGINT, SIG_IGN);
             $writer("worker2.startup");
         }, 'loop' => function () use($writer) {
             $writer("worker2.call");
         }, 'interval' => 1)));
         $builder->setLogger(new Logger('test', array($handler)))->setShutdownTimeout(3);
         $daemon = $builder->build();
         $daemon->setProcessName('testing');
         $daemon->run();
     });
     sleep(1);
     $start = time();
     $fork->kill(SIGQUIT);
     while (posix_kill($fork->getPid(), 0)) {
         pcntl_waitpid($fork->getPid(), $status, WNOHANG | WUNTRACED);
         usleep(100000);
     }
     $end = time();
     $diff = $end - $start;
     $this->assertTrue($diff >= 2 && $diff <= 4, 'Has been killed in shutdown interval');
     $content = file_get_contents($this->tempFile);
     $this->assertSame(1, preg_match_all('/worker1\\.call/', $content));
     $this->assertSame(1, preg_match_all('/worker2\\.startup/', $content));
     $calls = preg_match_all('/worker2\\.call/', $content);
     $this->assertTrue($calls >= 3 && $calls <= 5, 'Expected amount of worker2 calls');
 }

 public static function receive_sqs_for_multi()
 {
     $t1 = microtime(true);
     $pcount = 3;
     $pstack = array();
     for ($i = 1; $i <= $pcount; $i++) {
         $pid = pcntl_fork();
         if ($pid == -1) {
             die('fork できません');
         } else {
             if ($pid) {
                 // 親プロセスの場合
                 $pstack[$pid] = true;
                 if (count($pstack) >= $pcount) {
                     unset($pstack[pcntl_waitpid(-1, $status, WUNTRACED)]);
                 }
             } else {
                 sleep(1);
                 self::receive_sqs_message();
                 exit;
                 //処理が終わったらexitする。
             }
         }
     }
     //先に処理が進んでしまうので待つ
     while (count($pstack) > 0) {
         unset($pstack[pcntl_waitpid(-1, $status, WUNTRACED)]);
     }
     $t2 = microtime(true);
     $process_time = $t2 - $t1;
     \Cli::write("Process time = " . $process_time);
 }

function signal_handler($signo)
{
    echo "child process is ending... signal number is {$signo}\n";
    while (($pid = pcntl_waitpid(-1, $stat, WNOHANG)) > 0) {
        sprintf("child %d terminated.\n", $pid);
    }
}

 /**
 * Process in Parallel.
 *
 * Run a function (with no return result) on each item in an array in parallel.
 * Note: This function is only useful if order is not important, and you don't
 * need any return values from the function (i.e. no inter-process communication).
 *
 * @param mixed   $func  A closure function to apply to each item in parallel.
 * @param array   $arr   The array to apply function to.
 * @param integer $procs Number of processes to run in parallel.
 *
 * @return void
 * 
 *
  example to run :
 
  $makeDir = function($a) {
  shell_exec('mkdir '.shellescapearg($a));
  }
 
  // An array to process
  $dirnames = array('a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k');
 
  // Run the process in parallel.
  Multithread::processParallel($makeDir, $dirnames, 8);
 * 
 */
 public static function processParallel($func, array $arr, $procs = NULL)
 {
     // to improve take task 5 by 5 and wait last of the group of 5 !
     if (empty($procs)) {
         $procs = Cpu::getCpuCores();
     }
     // Break array up into $procs chunks.
     $chunks = array_chunk($arr, ceil(count($arr) / $procs));
     $pid = -1;
     $children = array();
     foreach ($chunks as $items) {
         $pid = pcntl_fork();
         if ($pid === -1) {
             die('could not fork');
         } else {
             if ($pid === 0) {
                 // We are the child process. Pass a chunk of items to process.
                 array_walk($items, $func);
                 exit(0);
             } else {
                 // We are the parent.
                 $children[] = $pid;
             }
         }
     }
     // Wait for children to finish.
     foreach ($children as $pid) {
         // We are still the parent.
         pcntl_waitpid($pid, $status);
     }
 }

 protected function scryed($total, $workers = 8, array $args)
 {
     $quiet = $this->option('quiet');
     $pids = [];
     for ($i = 0; $i < $workers; $i++) {
         $pids[$i] = pcntl_fork();
         switch ($pids[$i]) {
             case -1:
                 echo "fork error : {$i} \r\n";
                 exit;
             case 0:
                 $limit = floor($total / $workers);
                 $offset = $i * $limit;
                 if ($i == $workers - 1) {
                     $limit = $total - $offset;
                 }
                 $this->info(">>> 一个子进程已开启  |   剩 " . ($workers - $i - 1) . " 个  |  pid = " . getmypid() . " | --limit = {$limit}  |  --offset = {$offset}");
                 sleep(2);
                 // 这个sleep仅为看清上面的info，可删
                 array_push($args, "--limit={$limit}", "--offset={$offset}");
                 $quiet && array_push($args, '--quiet');
                 pcntl_exec('/usr/bin/php', $args, []);
                 // exec("/usr/bin/php5 artisan crawler:model autohome --sych-model=false --offset={$offset} --limit={$limit}");
                 // pcntl_exec 与 exec 同样可以执行命令。区别是exec 不会主动输出到shell控制台中
                 exit;
             default:
                 break;
         }
     }
     foreach ($pids as $pid) {
         $pid && pcntl_waitpid($pid, $status);
     }
 }

 public function child_signal_handler($signo, $pid = null, $status = null)
 {
     //If no pid is provided, that means we're getting the signal from the system.  Let's figure out
     //which child process ended
     if (!$pid) {
         $pid = pcntl_waitpid(-1, $status, WNOHANG);
     }
     //Make sure we get all of the exited children
     while ($pid > 0) {
         if ($pid && isset($this->current_jobs[$pid])) {
             $exit_code = pcntl_wexitstatus($status);
             if ($exit_code != 0) {
                 echo "{$pid} exited with status " . $exit_code . "\n";
             }
             unset($this->current_jobs[$pid]);
         } else {
             if ($pid) {
                 //Oh no, our job has finished before this parent process could even note that it had been launched!
                 echo "..... Adding {$pid} to the signal queue ..... \n";
                 $this->signal_queue[$pid] = $status;
             }
         }
         $pid = pcntl_waitpid(-1, $status, WNOHANG);
     }
     return true;
 }

 /**
  * run a callback in parrallel thread
  *
  * @param callable $callback
  * @return void
  */
 public function run(\Closure $callback = null)
 {
     $callback = $callback ?: function () {
     };
     $children = array();
     while (!empty($this->works)) {
         $items = array_pop($this->works);
         $pid = pcntl_fork();
         if (-1 == $pid) {
             throw new \RuntimeException('無法使用子程序!');
         } elseif ($pid) {
             //子程序start
             $children[] = $pid;
             $this->onStart($pid);
         } else {
             try {
                 //子程序處理 callback
                 $callback($items);
                 exit(0);
             } catch (\Exception $e) {
                 exit($e->getCode());
             }
         }
     }
     if ($children) {
         foreach ($children as $child) {
             $res = pcntl_waitpid($child, $status);
             if ($res == -1 || $res > 0) {
                 //子程序end
                 $status = pcntl_wexitstatus($status);
                 $this->onExit($child, $status);
             }
         }
     }
 }

 public function init()
 {
     $this->setGroupAndUser();
     if ($this->isDaemonActive()) {
         echo 'The Daemon is running already!' . "\n";
         exit;
     }
     $this->startDaemon();
     while (!$this->stopServer) {
         if (count($this->childProcesses, COUNT_RECURSIVE) - count($this->childProcesses) < $this->maxChild) {
             $this->startChildProcess();
         }
         usleep(self::DELAY);
         while ($signaled_pid = pcntl_waitpid(-1, $status, WNOHANG)) {
             if ($signaled_pid == -1) {
                 $this->childProcesses = [];
                 break;
             } else {
                 foreach ($this->childProcesses as $key => $childPids) {
                     if (is_array($childPids)) {
                         foreach ($childPids as $keyPid => $value) {
                             if ($value == $signaled_pid) {
                                 unset($this->childProcesses[$key][$keyPid]);
                             }
                         }
                     }
                 }
             }
         }
     }
 }

 private function process()
 {
     $messageType = NULL;
     $messageMaxSize = 1024;
     while (TRUE) {
         if (count($this->childs) < $this->max) {
             echo count($this->childs) . " ";
             if (msg_receive($this->queue, QUEUE_TYPE_START, $messageType, $messageMaxSize, $this->message)) {
                 $pid = pcntl_fork();
                 if ($pid == -1) {
                     die('could not fork' . PHP_EOL);
                 } else {
                     if ($pid) {
                         $this->childs[$pid] = TRUE;
                         $messageType = NULL;
                         $this->message = NULL;
                     } else {
                         sleep(3);
                         $this->complete($messageType, $this->message);
                         exit;
                     }
                 }
                 foreach ($this->childs as $pid => $value) {
                     if (pcntl_waitpid($pid, $status, WNOHANG)) {
                         if (pcntl_wifexited($status)) {
                             unset($this->childs[$pid]);
                         }
                     }
                 }
             }
         }
         sleep(1);
     }
 }

 public function testDealerRep()
 {
     $pids[] = $this->forkRepWorker();
     $pids[] = $this->forkRepWorker();
     $loop = new StreamSelectLoop();
     $context = new Context($loop);
     $dealer = $context->getSocket(\ZMQ::SOCKET_DEALER);
     $dealer->bind('ipc://test2.ipc');
     sleep(1);
     $msgs = array();
     $dealer->on('message', function ($msg) use(&$msgs) {
         $msgs[] = $msg;
     });
     $dealer->send(array('A', '', 'foo'));
     $dealer->send(array('B', '', 'bar'));
     $loop->addTimer(1, function () use($loop) {
         $loop->stop();
     });
     $loop->run();
     foreach ($pids as $pid) {
         pcntl_waitpid($pid, $status, WUNTRACED);
     }
     $this->assertCount(2, $msgs);
     $this->assertContains(array('A', '', 'foobar'), $msgs);
     $this->assertContains(array('B', '', 'barbar'), $msgs);
 }

 protected function sendShutdown()
 {
     $ret = $this->controller->send("SHUTDOWN");
     foreach ($this->getWorkers() as $worker) {
         pcntl_waitpid($worker->getPid(), $status);
     }
 }

 public function waitForThreads()
 {
     foreach ($this->pids as $pid) {
         pcntl_waitpid($pid, $status);
     }
     $this->pids = array();
 }