refactor(worker): 重构工作进程管理实现

- 将 WorkerCommand 从继承 Server 改为继承 Command
- 添加自定义 Worker 类封装 Workerman 功能
- 实现队列工作进程，支持内存监控和自动重启
- 实现定时任务工作进程，集成 Crontab 功能
- 更新事件触发机制，传递 Worker 实例参数
- 添加输入参数转换为 argv 的辅助方法
- 集成 Workerman 工作进程生命周期管理

composer.json

@@ -10,7 +10,9 @@
     "php": ">=8.3",
     "ext-mbstring": "*",
     "robmorgan/phinx": "^0.16.10",
-    "six-shop/core": ">=0.6 <1.0"
+    "six-shop/core": ">=0.6 <1.0",
+    "workerman/crontab": "^1.0.7",
+    "workerman/workerman": "^5.2.2"
   },
   "authors": [
     {

src/Command/CrontabCommand.php

@@ -59,8 +59,9 @@ class CrontabCommand extends Command
             }
         };
 
-        $worker->setOnWorkerStart(function () {
-            Event::trigger(CrontabWorkerStartEvent::class);
+        $worker->setOnWorkerStart(function (Worker $worker) {
+            $worker->name = 'crontab';
+            Event::trigger(CrontabWorkerStartEvent::class, $worker);
         });
         Worker::runAll();
     }

src/Command/WorkerCommand.php

@@ -4,14 +4,24 @@ declare(strict_types=1);
 
 namespace SixShop\System\Command;
 
+use SixShop\System\Event\CrontabWorkerStartEvent;
+use think\console\Command;
+use think\console\Input;
 use think\console\input\Argument;
 use think\console\input\Option;
-use think\worker\command\Server;
-use think\worker\Manager;
+use think\console\Output;
+use SixShop\System\Workerman\Worker as SixShopWorker;
+use Workerman\Crontab\Crontab;
+use Workerman\Timer;
 use Workerman\Worker;
 
-class WorkerCommand extends Server
+class WorkerCommand extends Command
 {
+    public function __construct(private \think\queue\Worker $queueWorker)
+    {
+        parent::__construct();
+    }
+
     public function configure(): void
     {
         $this->setName('system:worker')
@@ -21,67 +31,106 @@ class WorkerCommand extends Server
             ->setDescription('Workerman Server for ThinkPHP (custom entry)');
     }
 
-    public function handle(Manager $manager)
+
+    protected function execute(Input $input, Output $output): void
     {
-        $action = $this->input->getArgument('action') ?: 'start';
-        $grace = $this->input->getOption('grace');
+        SixShopWorker::setArgv($this->inputToArgv($input));
+        SixShopWorker::setApp($this->app);
+        $this->addCronWorker();
+        $this->addQueueWorker();
+        SixShopWorker::runAll();
+    }
 
-        // 启动流程，直接使用框架原生启动逻辑
-        if ($action === 'start') {
-            if ($grace) {
-                putenv('WORKER_GRACE=1');
-            }
-            $manager->start();
-            return;
+    /**
+     * @param Input $input
+     * @return array
+     */
+    private function inputToArgv(Input $input): array
+    {
+        $argv = [PHP_BINARY, 'worker', $input->getArgument('action')];
+        $daemon = $input->getOption('daemon');
+        if ($daemon) {
+            $argv[] = '-d';
         }
-
-        // 1. 重置全局$argv，绕过父类私有static::$command，底层getArgv正常识别指令
-        global $argv;
-        $argv = [
-            PHP_BINARY,
-            'think',
-            'worker',
-            $action
-        ];
+        $grace = $input->getOption('grace');
         if ($grace) {
             $argv[] = '-g';
         }
+        return $argv;
+    }
 
-        $pidFile = runtime_path() . 'worker.pid';
-        $masterPid = file_exists($pidFile) ? (int)file_get_contents($pidFile) : 0;
-        // 检测进程是否运行
-        if ($masterPid <= 0 || !posix_kill($masterPid, 0)) {
-            $this->output->error('Worker is not running');
-            return;
-        }
+    /**
+     * @return void
+     */
+    private function addCronWorker(): void
+    {
+        SixShopWorker::addWorker(
+            onWorkerStart: function (Worker $worker) {
+                $this->app->event->trigger(CrontabWorkerStartEvent::class, $worker);
+            },
+            name: 'crontab',
+            count: 1,
+            onWorkerStop: function (Worker $worker) {
+                foreach (Crontab::getAll() as $crontab) {
+                    /** @var Crontab $crontab */
+                    $crontab->destroy();
+                }
+            }
+        );
+    }
+    /**
+     * @return void
+     */
+    private function addQueueWorker(): void
+    {
+        SixShopWorker::addWorker(
+            onWorkerStart: function (Worker $worker) {
+                $shouldExit = false;
+                $logger = $this->app->log;
+                $connection = $this->app->config->get('queue.default');
+                $queue = $this->app->config->get("queue.connections.{$connection}.queue", 'default');
+                $delay = 0;
+                $sleep = 2;
+                $tries = 3;
+                $memoryLimit = 128;  // 内存上限MB
+                $queueWorker = $this->queueWorker;
 
-        match ($action) {
-            // 查询类：调用Workerman底层原生打印表格逻辑
-            'status', 'connections' => $this->renderWorkerStatus(),
+                $run = function () use (&$run, &$shouldExit, $queueWorker, $connection, $queue, $delay, $sleep, $tries, $memoryLimit, $logger, $worker) {
+                    if ($shouldExit) {
+                        $logger->info(sprintf("[pid:%d] 收到退出标记，停止消费调度", $worker->pid));
+                        return;
+                    }
 
-            // 信号控制类
-            'stop' => posix_kill($masterPid, $grace ? SIGQUIT : SIGINT),
-            'reload' => posix_kill($masterPid, $grace ? SIGUSR2 : SIGUSR1),
-            'restart' => posix_kill($masterPid, $grace ? SIGQUIT : SIGINT),
+                    try {
+                        $queueWorker->runNextJob($connection, $queue, $delay, $sleep, $tries);
+                    } catch (\Throwable $e) {
+                        $logger->error(sprintf(
+                            "[pid:%d] 队列底层异常：%s \n堆栈：%s",
+                            $worker->pid,
+                            $e->getMessage(),
+                            $e->getTraceAsString()
+                        ));
+                    }
 
-            default => $this->output->error("Unknown action: {$action}"),
-        };
+                    $usedMb = memory_get_usage(true) / 1024 / 1024;
+                    if ($usedMb > $memoryLimit) {
+                        $logger->info(sprintf(
+                            "[pid:%d] 内存超限 %.2fMB > %dMB，进程标记退出自愈",
+                            $worker->pid,
+                            $usedMb,
+                            $memoryLimit
+                        ));
+                        $shouldExit = true;
+                        return;
+                    }
 
-        // 非查询指令输出提示
-        if (!in_array($action, ['status', 'connections'])) {
-            $this->output->info("{$action} signal sent to master PID: {$masterPid}");
-        }
-    }
+                    Timer::add(1, $run, [], false);
+                };
 
-    /**
-     * 复用Workerman底层 status / connections 原生输出逻辑
-     */
-    protected function renderWorkerStatus(): void
-    {
-        Worker::$pidFile = runtime_path() . 'worker.pid';
-        Worker::$statusFile = runtime_path() . 'worker.status';
-        Worker::$logFile = runtime_path() . 'worker.log';
-        // 底层自动读取重写后的$argv，输出进程/连接表格
-        Worker::runAll();
+                Timer::add(1, $run, [], false);
+            },
+            name: 'queue',
+            count: 1,
+        );
     }
 }

src/Event/CrontabWorkerStartEvent.php

@@ -4,6 +4,11 @@ declare(strict_types=1);
 
 namespace SixShop\System\Event;
 
+use think\worker\Worker;
+
 class CrontabWorkerStartEvent
 {
+    public function __construct(public Worker $worker)
+    {
+    }
 }

src/Hook/GatheringCrontabEventHook.php

@@ -15,9 +15,8 @@ use SixShop\System\Event\GetCronJobsEvent;
 use SixShop\System\ExtensionManager;
 use think\App;
 use think\facade\Log;
-use think\worker\Manager;
-use think\worker\Worker;
 use Workerman\Crontab\Crontab;
+use Workerman\Worker;
 
 use function SixShop\Core\extension_name_list;
 
@@ -59,22 +58,11 @@ class GatheringCrontabEventHook
         };
     }
 
-    /**
-     * think-worker 启动时注册 crontab 专用子进程
-     */
-    #[Hook('worker.init')]
-    public function registerCrontabWorker(Manager $manager): void
-    {
-        $manager->addWorker(function (Worker $worker) {
-            $this->app->event->trigger(CrontabWorkerStartEvent::class);
-        }, 'crontab', 1);
-    }
-
     /**
      * @throws \ReflectionException
      */
     #[Hook(CrontabWorkerStartEvent::class)]
-    public function onWorkerStart(): void
+    public function onWorkerStart(Worker $worker): void
     {
         $extensionManager = $this->app->make(ExtensionManager::class);
         $event = new GetCronJobsEvent();

src/Workerman/Worker.php

@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace SixShop\System\Workerman;
+
+use think\App;
+
+final class Worker extends \Workerman\Worker
+{
+    private static array $argv;
+
+    private static App $app;
+
+    protected static function init(): void
+    {
+        self::$pidFile = self::$app->getRootPath() . 'runtime/worker.pid';
+        self::$logFile = self::$app->getRootPath() . 'runtime/log/worker.log';
+        self::$statusFile = self::$app->getRootPath() . 'runtime/worker.status';
+        parent::init();
+    }
+
+    public static function addWorker(callable $onWorkerStart, $name = 'none', int $count = 1, ?callable $onWorkerStop = null): void
+    {
+        $worker = new self();
+        $worker->name = $name;
+        $worker->count = $count;
+        $worker->onWorkerStart = $onWorkerStart($worker);
+        $worker->onWorkerStop = $onWorkerStop ? $onWorkerStop($worker) : null;
+    }
+
+    public static function setApp(App $app): void
+    {
+        self::$app = $app;
+    }
+    public static function setArgv(array $argv): void
+    {
+        self::$argv = $argv;
+    }
+    public static function getArgv(): array
+    {
+        return self::$argv;
+    }
+}