快捷導(dǎo)航

詳解PHP swoole process的使用方法

更新時(shí)間：2017年08月26日 14:19:51 作者：invokermiracle

這篇文章主要為大家詳細(xì)介紹了詳解PHP swoole process的使用方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

引入背景：假如我們每天有10000個(gè)訂單生成，需要同步到倉儲(chǔ)系統(tǒng)中去，以前做法是開啟一個(gè)crontab去跑這些任務(wù)，但是發(fā)現(xiàn)總有感覺同步效率低，間隔時(shí)間都是分鐘級(jí)別的。

解決方案測(cè)試：我們將同步訂單的任務(wù)表添加一個(gè)hash作為key，作為分發(fā)條件，因?yàn)閙ysql中select如果做mod函數(shù)是用不到索引的，所以我們自己做隨機(jī)hash，但是務(wù)必不需要范圍太大，以免服務(wù)器資源不夠，方法是根據(jù)hashkey投放到不同的進(jìn)程中進(jìn)行同步，測(cè)試代碼如下

<?php
/**
 * Created by PhpStorm.
 * User: xujun
 * Date: 2017/8/26
 * Time: 9:37
 */
//假定需要處理的數(shù)據(jù)如下
class Process{
  public $mpid=0;
  public $max_precess=5;
  //代替從數(shù)據(jù)庫中讀取的內(nèi)容
  public $task = [
    ['uid'=>1,'uname'=>'bot','hash'=>1,'handle'=>'test'],
    ['uid'=>2,'uname'=>'bot1','hash'=>2,'handle'=>'test'],
    ['uid'=>3,'uname'=>'bot2','hash'=>3,'handle'=>'test'],
    ['uid'=>4,'uname'=>'bot3','hash'=>4,'handle'=>'test'],
    ['uid'=>2,'uname'=>'bot4','hash'=>2,'handle'=>'test'],
    ['uid'=>3,'uname'=>'bot5','hash'=>3,'handle'=>'test'],
    ['uid'=>4,'uname'=>'bot6','hash'=>1,'handle'=>'test'],
  ];
  public $works = [];
  public $swoole_table = NULL;
  //public $new_index=0;
  function test($index,$task){
    print_r("[".date('Y-m-d H:i:s')."]".'work-index:'.$index.'處理'.$task['uname'].'完成'.PHP_EOL);
  }

  public function __construct(){
    try {
      $this->swoole_table = new swoole_table(1024);
      $this->swoole_table->column('index', swoole_table::TYPE_INT);//用于父子進(jìn)程間數(shù)據(jù)交換
      $this->swoole_table->create();

      swoole_set_process_name(sprintf('php-ps:%s', 'master'));
      $this->mpid = posix_getpid();
      $this->run();
      $this->processWait();
    }catch (\Exception $e){
      die('ALL ERROR: '.$e->getMessage());
    }
  }

  public function run(){
    for ($i=0; $i < $this->max_precess; $i++) {
      $this->CreateProcess();
    }
  }

  private function getTask($index){
    $_return = [];
    foreach ($this->task as $v){
      if($v['hash']==$index){
        $_return[] = $v;
      }
    }
    return $_return;
  }

  public function CreateProcess($index=null){
    if(is_null($index)){//如果沒有指定了索引，新建的子進(jìn)程，開啟計(jì)數(shù)
      $index=$this->swoole_table->get('index');
      if($index === false){
        $index = 0;
      }else{
        $index = $index['index']+1;
      }
      print_r($index);
    }
    $this->swoole_table->set('index',array('index'=>$index));
    $process = new swoole_process(function(swoole_process $worker)use($index){

      swoole_set_process_name(sprintf('php-ps:%s',$index));
      $task = $this->getTask($index);
      foreach ($task as $v){
        call_user_func_array(array($this,$v['handle']),array($index,$v));
      }
      sleep(20);
    }, false, false);
    $pid=$process->start();

    $this->works[$index]=$pid;
    return $pid;
  }

  public function rebootProcess($ret){
    $pid=$ret['pid'];
    $index=array_search($pid, $this->works);
    if($index!==false){
      $index=intval($index);
      $new_pid=$this->CreateProcess($index);
      echo "rebootProcess: {$index}={$new_pid} Done\n";
      return;
    }
    throw new \Exception('rebootProcess Error: no pid');
  }

  public function processWait(){
    while(1) {
      if(count($this->works)){
        $ret = swoole_process::wait();
        if ($ret) {
          $this->rebootProcess($ret);
        }
      }else{
        break;
      }
    }
  }

}
$process = new Process();

這里代碼中，使用了swoole_table作為進(jìn)程間共享的內(nèi)存，為了分配index。以及當(dāng)進(jìn)程退出后，父進(jìn)程通過wait重新拉起該進(jìn)程任務(wù)。

測(cè)試截圖

進(jìn)程ps