自定义进程

EasySwoole中支持添加用户自定义的swoole process。

抽象父类

任何的自定义进程，都应该继承自EasySwoole\Core\Swoole\Process\AbstractProcess,
AbstractProcess实现代码如下：

  1. <?php
  2. /**
  3.  * Created by PhpStorm.
  4.  * User: yf
  5.  * Date: 2018/1/17
  6.  * Time: 上午11:28
  7.  */
  9. namespace EasySwoole\Core\Swoole\Process;
  12. use EasySwoole\Core\Swoole\Memory\TableManager;
  13. use EasySwoole\Core\Swoole\ServerManager;
  14. use EasySwoole\Core\Swoole\Time\Timer;
  15. use Swoole\Process;
  17. abstract class AbstractProcess
  18. {
  19.     private $swooleProcess;
  20.     private $processName;
  21.     private $async = null;
  22.     private $args = [];
  23.     function __construct(string $processName,$async = true,array $args)
  24.     {
  25.         $this->async = $async;
  26.         $this->args = $args;
  27.         $this->processName = $processName;
  28.         $this->swooleProcess = new \swoole_process([$this,'__start'],false,2);
  29.         ServerManager::getInstance()->getServer()->addProcess($this->swooleProcess);
  30.     }
  32.     public function getProcess():Process
  33.     {
  34.         return $this->swooleProcess;
  35.     }
  37.     /*
  38.      * 仅仅为了提示:在自定义进程中依旧可以使用定时器
  39.      */
  40.     public function addTick($ms,callable $call):?int
  41.     {
  42.         return Timer::loop(
  43.             $ms,$call
  44.         );
  45.     }
  47.     public function clearTick(int $timerId)
  48.     {
  49.         Timer::clear($timerId);
  50.     }
  52.     public function delay($ms,callable $call):?int
  53.     {
  54.         return Timer::delay(
  55.             $ms,$call
  56.         );
  57.     }
  59.     /*
  60.      * 服务启动后才能获得到pid
  61.      */
  62.     public function getPid():?int
  63.     {
  64.         if(isset($this->swooleProcess->pid)){
  65.             return $this->swooleProcess->pid;
  66.         }else{
  67.             $key = md5($this->processName);
  68.             $pid = TableManager::getInstance()->get('process_hash_map')->get($key);
  69.             if($pid){
  70.                 return $pid['pid'];
  71.             }else{
  72.                 return null;
  73.             }
  74.         }
  75.     }
  78.     function __start(Process $process)
  79.     {
  80.         if(PHP_OS != 'Darwin'){
  81.             $process->name($this->getProcessName());
  82.         }
  83.         TableManager::getInstance()->get('process_hash_map')->set(
  84.             md5($this->processName),['pid'=>$this->swooleProcess->pid]
  85.         );
  86.         ProcessManager::getInstance()->setProcess($this->getProcessName(),$this);
  87.         if (extension_loaded('pcntl')) {
  88.             pcntl_async_signals(true);
  89.         }
  90.         Process::signal(SIGTERM,function ()use($process){
  91.             $this->onShutDown();
  92.             TableManager::getInstance()->get('process_hash_map')->del(md5($this->processName));
  93.             swoole_event_del($process->pipe);
  94.             $this->swooleProcess->exit(0);
  95.         });
  96.         if($this->async){
  97.             swoole_event_add($this->swooleProcess->pipe, function(){
  98.                 $msg = $this->swooleProcess->read(64 * 1024);
  99.                 $this->onReceive($msg);
  100.             });
  101.         }
  102.         $this->run($this->swooleProcess);
  103.     }
  105.     public function getArgs():array
  106.     {
  107.         return $this->args;
  108.     }
  110.     public function getProcessName()
  111.     {
  112.         return $this->processName;
  113.     }
  115.     public abstract function run(Process $process);
  116.     public abstract function onShutDown();
  117.     public abstract function onReceive(string $str,...$args);
  119. }

进程管理器

ProcessManager，实现代码如下：

  1. namespace EasySwoole\Core\Swoole\Process;
  4. use EasySwoole\Core\AbstractInterface\Singleton;
  5. use EasySwoole\Core\Swoole\Memory\TableManager;
  6. use EasySwoole\Core\Swoole\ServerManager;
  7. use Swoole\Table;
  10. class ProcessManager
  11. {
  12.     use Singleton;
  13.     private $processList = [];
  15.     function __construct()
  16.     {
  17.         TableManager::getInstance()->add(
  18.             'process_hash_map',[
  19.                 'pid'=>[
  20.                     'type'=>Table::TYPE_INT,
  21.                     'size'=>10
  22.                 ]
  23.             ],256
  24.         );
  25.     }
  27.     public function addProcess(string $processName,string $processClass,$async = true,array $args = []):bool
  28.     {
  29.         if(ServerManager::getInstance()->isStart()){
  30.             trigger_error('you can not add a process after server start');
  31.             return false;
  32.         }
  33.         $key = md5($processName);
  34.         if(!isset($this->processList[$key])){
  35.             try{
  36.                 $process = new $processClass($processName,$async,$args);
  37.                 $this->processList[$key] = $process;
  38.                 return true;
  39.             }catch (\Throwable $throwable){
  40.                 trigger_error($throwable->getMessage().$throwable->getTraceAsString());
  41.                 return false;
  42.             }
  43.         }else{
  44.             trigger_error('you can not add the same name process : '.$processName);
  45.             return false;
  46.         }
  47.     }
  49.     public function getProcessByName(string $processName):?AbstractProcess
  50.     {
  51.         $key = md5($processName);
  52.         if(isset($this->processList[$key])){
  53.             return $this->processList[$key];
  54.         }else{
  55.             return null;
  56.         }
  57.     }
  60.     public function getProcessByPid(int $pid):?AbstractProcess
  61.     {
  62.         $table = TableManager::getInstance()->get('process_hash_map');
  63.         foreach ($table as $key => $item){
  64.             if($item['pid'] == $pid){
  65.                 return $this->processList[$key];
  66.             }
  67.         }
  68.         return null;
  69.     }
  72.     public function setProcess(string $processName,AbstractProcess $process)
  73.     {
  74.         $key = md5($processName);
  75.         $this->processList[$key] = $process;
  76.     }
  78.     public function reboot(string $processName):bool
  79.     {
  80.         $p = $this->getProcessByName($processName);
  81.         if($p){
  82.             \swoole_process::kill($p->getPid(),SIGTERM);
  83.             return true;
  84.         }else{
  85.             return false;
  86.         }
  87.     }
  89.     public function writeByProcessName(string $name,string $data):bool
  90.     {
  91.         $process = $this->getProcessByName($name);
  92.         if($process){
  93.             return (bool)$process->getProcess()->write($data);
  94.         }else{
  95.             return false;
  96.         }
  97.     }
  99.     public function readByProcessName(string $name,float $timeOut = 0.1):?string
  100.     {
  101.         $process = $this->getProcessByName($name);
  102.         if($process){
  103.             $process = $process->getProcess();
  104.             $read = array($process);
  105.             $write = [];
  106.             $error = [];
  107.             $ret = swoole_select($read, $write,$error, $timeOut);
  108.             if($ret){
  109.                 return $process->read(64 * 1024);
  110.             }else{
  111.                 return null;
  112.             }
  113.         }else{
  114.             return null;
  115.         }
  116.     }
  118. }

实例

我们以demo中的自定义进程例子来说明：

  1. namespace App\Process;
  2. use EasySwoole\Core\Swoole\Process\AbstractProcess;
  3. use Swoole\Process;
  4. class Test extends AbstractProcess
  5. {
  6.     //进程start的时候会执行的事件
  7.     public function run(Process $process)
  8.     {
  9.         // TODO: Implement run() method.
  10.         //添加进程内定时器
  11.         $this->addTick(2000,function (){
  12.             var_dump('this is '.$this->getProcessName().' process tick');
  13.         });
  14.     }
  15.     //当进程关闭的时候会执行该事件
  16.     public function onShutDown()
  17.     {
  18.         // TODO: Implement onShutDown() method.
  19.     }
  20.     //当有信息发给该进程的时候，会执行此进程
  21.     public function onReceive(string $str, ...$args)
  22.     {
  23.         // TODO: Implement onReceive() method.
  24.         var_dump('process rec'.$str);
  25.     }
  26. }

以上代码直达连接，
至于如何使用（测试），请见demo中的EasySwooleEvent.php