限制Worker进程异常退出后，10分钟内最多10次重新拉起，超过限制后不再重新拉起；Master进程名显示当前Master的启动文件名。

Author: lileiseven

CHANGELOG.md

@@ -3,6 +3,11 @@
 ## HEAD (Unreleased)
 _(none)_
 
+## 0.8.0 (2017-08-13)
+
+* 限制Worker进程异常退出后，10分钟内最多10次重新拉起，超过限制后不再重新拉起。
+* Master进程名显示当前Master的启动文件名。
+
 ## 0.7.5 (2017-08-11)
 
 * 修复 ForawrdWorker 未设 Logger 报错的问题。

bin/plumber

@@ -45,5 +45,5 @@ if (!$config instanceof \Pimple\Container ) {
     exit(2);
 }
 
-$plumber = new Plumber($config);
+$plumber = new Plumber($config, $configFile);
 $plumber->main($cmd);

composer.json

@@ -23,8 +23,9 @@
         "ext-swoole": ">=1.7.17",
         "pimple/pimple": "^3",
         "docopt/docopt": "^1",
-        "codeages/beanstalk-client" : "^0.1.4",
-        "monolog/monolog": "^1.22.0"
+        "codeages/beanstalk-client" : "^0.1",
+        "codeages/rate-limiter": "^0.2",
+        "monolog/monolog": "^1.22"
     },
     "bin": ["bin/plumber"],
     "require-dev": {

example/Example2Worker.php

@@ -10,7 +10,8 @@ class Example2Worker implements IWorker
 
     public function execute($data)
     {
-        $this->logger->info("I'm example 2 worker.");
+        sleep(1);
+        throw new \RuntimeException("exception from example2 worker.");
 
         return array('code' => IWorker::FINISH);
     }

example/put_message.php

@@ -13,7 +13,7 @@
 $beanstalk = new BeanstalkClient($config);
 
 $beanstalk->connect();
-$beanstalk->useTube('ExampleForward');
+$beanstalk->useTube('Example2');
 
 $i=0;
 

src/ContainerAwareTrait.php

@@ -1,4 +1,5 @@
 <?php
+
 namespace Codeages\Plumber;
 
 use Pimple\Container;

src/ForwardWorker.php

@@ -2,7 +2,6 @@
 
 namespace Codeages\Plumber;
 
-use Psr\Log\LoggerInterface;
 use Codeages\Beanstalk\Client as BeanstalkClient;
 use Codeages\Beanstalk\ClientProxy as BeanstalkClientProxy;
 

src/IWorker.php

@@ -1,4 +1,5 @@
 <?php
+
 namespace Codeages\Plumber;
 
 use Pimple\Container;

src/Plumber.php

@@ -2,12 +2,13 @@
 
 namespace Codeages\Plumber;
 
+use swoole_process;
+use Psr\Log\LoggerInterface;
 use Pimple\Container;
 use Monolog\Logger;
 use Monolog\Handler\StreamHandler;
 use Monolog\ErrorHandler;
-use Psr\Log\LoggerInterface;
-use swoole_process;
+use Codeages\RateLimiter\RateLimiter;
 
 class Plumber
 {
@@ -16,8 +17,10 @@ class Plumber
      */
     protected $container;
 
+    protected $configFilePath;
+
     /**
-     * @var boolean
+     * @var bool
      */
     protected $daemon;
 
@@ -28,15 +31,27 @@ class Plumber
      */
     protected $logger;
 
+    /**
+     * @var RateLimiter
+     */
+    protected $limiter;
+
     const ALREADY_RUNNING_ERROR = 1;
 
     const LOCK_PROCESS_ERROR = 2;
 
-    public function __construct(Container $container)
+    public function __construct(Container $container, $configFilePath)
     {
         $container['run_flag'] = new SharedRunFlag();
         $this->locker = new ProcessLocker($container['pid_path']);
+        $this->limiter = new RateLimiter(
+            'process_recreate',
+            10,
+            600,
+            new \Codeages\RateLimiter\Storage\ArrayStorage()
+        );
         $this->container = $container;
+        $this->configFilePath = $configFilePath;
     }
 
     public function main($op)
@@ -70,7 +85,7 @@ protected function start($daemon = true)
             exit(self::LOCK_PROCESS_ERROR);
         }
 
-        swoole_set_process_name('plumber: master');
+        swoole_set_process_name(sprintf('plumber: master (%s)', $this->configFilePath));
 
         $logger = new Logger('plumber');
         if ($daemon) {
@@ -81,12 +96,12 @@ protected function start($daemon = true)
         ErrorHandler::register($logger);
         $this->container['logger'] = $this->logger = $logger;
 
-        $this->workers = $this->createWorkers();
+        $this->workers = $this->startWorkers();
         $this->registerSignal();
 
         foreach ($this->workers as $worker) {
-            swoole_event_add($worker->pipe, function ($pipe) use ($worker, $logger) {
-                $logger->info("read from worker:".$worker->read());
+            swoole_event_add($worker['process']->pipe, function ($pipe) use ($worker, $logger) {
+                $logger->info('read from worker:'.$worker['process']->read());
             });
         }
 
@@ -106,6 +121,7 @@ protected function stop()
         $pid = $this->locker->getId();
         if (empty($pid)) {
             echo "plumber is not running.\n";
+
             return;
         }
 
@@ -132,20 +148,23 @@ protected function restart()
     /**
      * 创建队列的监听器.
      */
-    private function createWorkers()
+    private function startWorkers()
     {
         $workers = [];
         foreach ($this->container['tubes'] as $name => $options) {
             for ($i = 0; $i < $options['worker_num']; ++$i) {
-                $worker = $this->createWorker($name);
-                $workers[$worker->pid] = $worker;
+                $worker = $this->startWorker($name);
+                $workers[$worker->pid] = array(
+                    'tube' => $name,
+                    'process' => $worker,
+                );
             }
         }
 
         return $workers;
     }
 
-    private function createWorker($queueName)
+    private function startWorker($queueName)
     {
         $worker = new \swoole_process(function ($process) use ($queueName) {
             $process->name("plumber: queue `{$queueName}` worker");
@@ -172,18 +191,26 @@ private function registerSignal()
                 }
 
                 if ($this->container['run_flag']->isRuning()) {
-                    $newPid = $this->workers[$ret['pid']]->start();
-                    $this->workers[$newPid] = $this->workers[$ret['pid']];
+                    $worker = $this->workers[$ret['pid']];
                     unset($this->workers[$ret['pid']]);
-                    $this->logger->notice("process #{$ret['pid']} exited, #{$newPid} is recreated.", $ret);
+
+                    $remainTimes = $this->limiter->check($worker['tube']);
+                    if ($remainTimes > 0) {
+                        $newPid = $worker['process']->start();
+                        $this->workers[$newPid] = $worker;
+                        $this->logger->notice("tube {$worker['tube']} process #{$ret['pid']} exited, #{$newPid} is recreated, remain {$remainTimes} recreated times. .", $ret);
+                    } else {
+                        $this->logger->notice("tube {$worker['tube']} process #{$ret['pid']} exited, reached max recreated times.", $ret);
+                    }
                 } else {
                     unset($this->workers[$ret['pid']]);
                     $this->logger->info("process #{$ret['pid']} exited.", $ret);
-                    if (empty($this->workers)) {
-                        $this->locker->release();
-                        $this->logger->info("stoped.");
-                        swoole_event_exit();
-                    }
+                }
+
+                if (empty($this->workers)) {
+                    $this->locker->release();
+                    $this->logger->info('stoped.');
+                    swoole_event_exit();
                 }
             }
         });

src/ProcessLocker.php

@@ -20,6 +20,7 @@ public function lock($id)
         }
         $id = intval($id);
         file_put_contents($this->path, $id);
+
         return true;
     }
 

src/SharedRunFlag.php

@@ -1,4 +1,5 @@
 <?php
+
 namespace Codeages\Plumber;
 
 use swoole_atomic;

src/WorkerProcess.php

@@ -52,9 +52,9 @@ public function run()
             try {
                 $result = $executor->execute($job);
             } catch (\Exception $e) {
-                $message = sprintf('tube({$tubeName}, #%d): execute job #%d exception, `%s`', $process->pid, $job['id'], $e->getMessage());
+                $message = sprintf('tube(%s, #%d): execute job #%d exception, `%s`', $this->tubeName, $process->pid, $job['id'], $e->getMessage());
                 $logger->error($message, $job);
-                continue;
+                throw $e;
             }
 
             $code = is_array($result) ? $result['code'] : $result;
@@ -170,13 +170,15 @@ private function retryJob($job, $result)
         $stats = $queue->statsJob($job['id']);
         if ($stats === false) {
             $logger->error("tube({$tubeName}, #{$process->pid}): job #{$job['id']} get stats failed, in retry executed.", $job);
+
             return;
         }
 
         $logger->info("tube({$tubeName}, #{$process->pid}): job #{$job['id']} retry {$message['__retry']} times.");
         $deleted = $queue->delete($job['id']);
         if (!$deleted) {
             $logger->error("tube({$tubeName}, #{$process->pid}): job #{$job['id']} delete failed, in retry executed.", $job);
+
             return;
         }
 
@@ -204,16 +206,18 @@ private function buryJob($job, $result)
         $stats = $queue->statsJob($job['id']);
         if ($stats === false) {
             $logger->error("tube({$tubeName}, #{$process->pid}): job #{$job['id']} get stats failed, in bury executed.", $job);
+
             return;
         }
 
         $pri = isset($result['pri']) ? $result['pri'] : $stats['pri'];
         $burried = $queue->bury($job['id'], $pri);
         if ($burried === false) {
             $logger->error("tube({$tubeName}, #{$process->pid}): job #{$job['id']} bury failed", $job);
+
             return;
         }
 
-        $logger->info("tube({$tubeName}, #{$process->pid}): job #{$job['id']} buried.");
+        $logger->notice("tube({$tubeName}, #{$process->pid}): job #{$job['id']} buried.");
     }
 }