debug

code/extend/microserver/ProcessSanitizer.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+use think\facade\Cache;
+use think\facade\Db;
+
+class ProcessSanitizer
+{
+    public static function reloadRedis(): void
+    {
+        try {
+            $redis = Cache::store('redis')->handler();
+            if ($redis instanceof \Redis) {
+                $redis->close();
+            }
+        } catch (\Throwable) {
+        }
+    }
+
+    public static function reloadDb(): void
+    {
+        try {
+            Db::disconnect();
+        } catch (\Throwable) {
+        }
+    }
+
+    public static function destructConnectSource(): void
+    {
+        self::reloadRedis();
+        self::reloadDb();
+    }
+}

code/extend/microserver/QueueManage.php

@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+use think\facade\Cache;
+
+class QueueManage
+{
+    private static array $instances = [];
+
+    private int $index;
+
+    private string $queueName;
+
+    public static function getInstance(int $intIndex = 1): self
+    {
+        if (!isset(self::$instances[$intIndex])) {
+            self::$instances[$intIndex] = new self($intIndex);
+        }
+
+        return self::$instances[$intIndex];
+    }
+
+    public function __construct(int $intIndex = 1)
+    {
+        $this->index = $intIndex;
+        $this->queueName = (string) config("task.queue.{$intIndex}.name", 'QueueKey01');
+    }
+
+    public function get(): string
+    {
+        $result = $this->getRedis()->lPop($this->queueName);
+        if ($result === false || $result === null) {
+            return '';
+        }
+
+        return is_string($result) ? $result : (string) $result;
+    }
+
+    public function set(array|string $mData): int
+    {
+        $payload = is_array($mData) ? json_encode($mData, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) : (string) $mData;
+        if ($payload === false || $payload === '') {
+            return 0;
+        }
+
+        return (int) $this->getRedis()->rPush($this->queueName, $payload);
+    }
+
+    private function getRedis(): \Redis
+    {
+        /** @var \Redis $redis */
+        $redis = Cache::store('redis')->handler();
+        return $redis;
+    }
+}

code/extend/microserver/Rout.php

@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+class Rout
+{
+    public static function httpDispense(...$args): void
+    {
+    }
+
+    public static function tcpDispense(...$args): void
+    {
+    }
+}

code/extend/microserver/ScheduledTasks.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+class ScheduledTasks
+{
+    public static function hander(array $arrTask): void
+    {
+        $arrCallback = $arrTask['callback'] ?? null;
+        if (!is_array($arrCallback) || count($arrCallback) < 2) {
+            return;
+        }
+
+        try {
+            call_user_func($arrCallback, $arrTask['param'] ?? []);
+        } catch (\Throwable $Throwable) {
+            ServerManage::appendLog(
+                sprintf(
+                    'Scheduled task failed: %s in %s:%d',
+                    $Throwable->getMessage(),
+                    $Throwable->getFile(),
+                    $Throwable->getLine()
+                )
+            );
+        }
+    }
+}

code/extend/microserver/ServerConsole.php

@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+use think\console\Command;
+use think\console\Input;
+use think\console\input\Argument;
+use think\console\input\Option;
+use think\console\Output;
+
+class ServerConsole extends Command
+{
+    protected function configure()
+    {
+        $this->setName('task')
+            ->setDescription('run task service')
+            ->setHelp('php think task <action> [-d] [-f] run server')
+            ->addArgument('action', Argument::REQUIRED, 'action: (start|stop|restart|status)')
+            ->addOption('daemonize', 'd', Option::VALUE_NONE, 'daemonize start service')
+            ->addOption('forcibly', 'f', Option::VALUE_NONE, 'ignore user forcibly start service');
+    }
+
+    protected function execute(Input $input, Output $output): int
+    {
+        $strAction = strtolower((string) $input->getArgument('action'));
+        if (!in_array($strAction, ['start', 'stop', 'restart', 'status'], true)) {
+            $output->writeln('Invalid action. Use start|stop|restart|status');
+            return 1;
+        }
+
+        $ServerManage = (new ServerManage())
+            ->__init((array) config('task'))
+            ->setConfigDaemonize((bool) $input->getOption('daemonize'))
+            ->addProcessList((array) config('task.process_pool', []))
+            ->addScheduledTasks((array) config('task.scheduled_tasks_pool', []))
+            ->createService();
+
+        return match ($strAction) {
+            'start' => $ServerManage->start(),
+            'stop' => $ServerManage->stop(),
+            'restart' => $ServerManage->restart(),
+            'status' => $ServerManage->status(),
+        };
+    }
+}

code/extend/microserver/ServerCore.php

@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+class ServerCore
+{
+    private static ?self $obj = null;
+
+    private array $arrConfig = [];
+
+    public static function instance(array $arrConfig = []): self
+    {
+        if (self::$obj === null) {
+            self::$obj = new self($arrConfig);
+        }
+
+        return self::$obj;
+    }
+
+    public function __construct(array $arrConfig = [])
+    {
+        $this->arrConfig = $arrConfig;
+    }
+
+    public function serHttpConfig(): array
+    {
+        return $this->arrConfig;
+    }
+
+    public function filterConfig(array $arrConfig = []): array
+    {
+        return $arrConfig;
+    }
+
+    public function serTcpConfig(): array
+    {
+        return $this->arrConfig;
+    }
+
+    public function createServer(): null
+    {
+        return null;
+    }
+
+    public function getServer(): null
+    {
+        return null;
+    }
+
+    public function onTask(...$args): void
+    {
+    }
+
+    public function onFinish(...$args): void
+    {
+    }
+}

code/extend/microserver/ServerManage.php

@@ -0,0 +1,348 @@
+<?php
+
+declare(strict_types=1);
+
+namespace microserver;
+
+use Swoole\Process;
+
+class ServerManage
+{
+    private array $arrDynamicConfig = [];
+
+    private array $processPool = [];
+
+    private array $scheduledTasks = [];
+
+    private bool $daemonize = false;
+
+    private bool $running = true;
+
+    private array $childDefinitions = [];
+
+    private array $childPids = [];
+
+    public function __init(array $arrDynamicConfig = []): self
+    {
+        $this->arrDynamicConfig = $arrDynamicConfig;
+        return $this;
+    }
+
+    public function setConfigDaemonize(bool $boolDaemonize): self
+    {
+        $this->daemonize = $boolDaemonize;
+        return $this;
+    }
+
+    public function getPid(): int
+    {
+        $strPidFile = (string) ($this->arrDynamicConfig['service']['pid_file'] ?? '');
+        if ($strPidFile === '' || !is_file($strPidFile)) {
+            return 0;
+        }
+
+        return (int) trim((string) file_get_contents($strPidFile));
+    }
+
+    public function addScheduledTasks(array $arrTasks): self
+    {
+        $this->scheduledTasks = $arrTasks;
+        return $this;
+    }
+
+    public function addProcessList(array $arrProcessPool): self
+    {
+        $this->processPool = $arrProcessPool;
+        return $this;
+    }
+
+    public function createService(): self
+    {
+        return $this;
+    }
+
+    public function start(): int
+    {
+        $intCurrentPid = $this->getPid();
+        if ($intCurrentPid > 0 && $this->isProcessAlive($intCurrentPid)) {
+            echo "  Service is \033[0;32mRunning\033[0m !  " . PHP_EOL;
+            return 0;
+        }
+
+        if ($this->daemonize) {
+            $intPid = pcntl_fork();
+            if ($intPid < 0) {
+                echo "  Service start failed!  " . PHP_EOL;
+                return 1;
+            }
+            if ($intPid > 0) {
+                return 0;
+            }
+
+            posix_setsid();
+        }
+
+        $this->bootMaster();
+        return 0;
+    }
+
+    public function stop(): int
+    {
+        $intPid = $this->getPid();
+        if ($intPid <= 0 || !$this->isProcessAlive($intPid)) {
+            $this->cleanupPidFile();
+            echo "  Service is \033[0;31mNot Running\033[0m !  " . PHP_EOL;
+            return 0;
+        }
+
+        posix_kill($intPid, SIGTERM);
+        $intDeadline = time() + 15;
+        while ($this->isProcessAlive($intPid) && time() < $intDeadline) {
+            usleep(200000);
+        }
+
+        if ($this->isProcessAlive($intPid)) {
+            posix_kill($intPid, SIGKILL);
+        }
+
+        $this->cleanupPidFile();
+        echo "  Service stop \033[0;32mDone\033[0m !  " . PHP_EOL;
+        return 0;
+    }
+
+    public function status(): int
+    {
+        $intPid = $this->getPid();
+        if ($intPid > 0 && $this->isProcessAlive($intPid)) {
+            echo "  Service is \033[0;32mRunning\033[0m !  " . PHP_EOL;
+            return 0;
+        }
+
+        echo "  Service is \033[0;31mNot Running\033[0m !  " . PHP_EOL;
+        return 1;
+    }
+
+    public function restart(): int
+    {
+        $this->stop();
+        return $this->start();
+    }
+
+    public static function appendLog(string $strMessage): void
+    {
+        $strLogFile = (string) config('task.service.log_file', runtime_path() . 'task.log');
+        $strDir = dirname($strLogFile);
+        if (!is_dir($strDir)) {
+            @mkdir($strDir, 0777, true);
+        }
+        @file_put_contents($strLogFile, '[' . date('Y-m-d H:i:s') . '] ' . $strMessage . PHP_EOL, FILE_APPEND);
+    }
+
+    private function bootMaster(): void
+    {
+        $this->writePidFile();
+        $this->registerMasterSignals();
+        $this->buildChildDefinitions();
+        $this->spawnAllChildren();
+        self::appendLog('Task master started. pid=' . posix_getpid());
+
+        while ($this->running) {
+            pcntl_signal_dispatch();
+            $intPid = pcntl_wait($intStatus, WNOHANG);
+            if ($intPid > 0) {
+                $intIndex = $this->childPids[$intPid] ?? null;
+                unset($this->childPids[$intPid]);
+                if ($this->running && $intIndex !== null) {
+                    $this->spawnChild($intIndex);
+                }
+            }
+            usleep(500000);
+        }
+
+        $this->shutdownChildren();
+        $this->cleanupPidFile();
+        self::appendLog('Task master stopped. pid=' . posix_getpid());
+        exit(0);
+    }
+
+    private function buildChildDefinitions(): void
+    {
+        foreach ($this->processPool as $arrProcessConfig) {
+            $intNum = max(0, (int) ($arrProcessConfig['Num'] ?? 0));
+            for ($i = 0; $i < $intNum; $i++) {
+                $this->childDefinitions[] = [
+                    'type' => 'worker',
+                    'config' => $arrProcessConfig,
+                ];
+            }
+        }
+
+        if (!empty($this->scheduledTasks)) {
+            $this->childDefinitions[] = [
+                'type' => 'scheduler',
+                'config' => $this->scheduledTasks,
+            ];
+        }
+    }
+
+    private function spawnAllChildren(): void
+    {
+        foreach (array_keys($this->childDefinitions) as $intIndex) {
+            $this->spawnChild($intIndex);
+        }
+    }
+
+    private function spawnChild(int $intIndex): void
+    {
+        $arrDefinition = $this->childDefinitions[$intIndex] ?? null;
+        if (!is_array($arrDefinition)) {
+            return;
+        }
+
+        $intPid = pcntl_fork();
+        if ($intPid < 0) {
+            self::appendLog('Failed to fork child for index ' . $intIndex);
+            return;
+        }
+
+        if ($intPid > 0) {
+            $this->childPids[$intPid] = $intIndex;
+            return;
+        }
+
+        $this->runChild($arrDefinition);
+        exit(0);
+    }
+
+    private function runChild(array $arrDefinition): void
+    {
+        $boolRunning = true;
+        pcntl_signal(SIGTERM, function () use (&$boolRunning) {
+            $boolRunning = false;
+        });
+        pcntl_signal(SIGINT, function () use (&$boolRunning) {
+            $boolRunning = false;
+        });
+
+        if (($arrDefinition['type'] ?? '') === 'scheduler') {
+            $this->runSchedulerLoop($arrDefinition['config'], $boolRunning);
+            return;
+        }
+
+        $this->runWorkerLoop($arrDefinition['config'], $boolRunning);
+    }
+
+    private function runWorkerLoop(array $arrProcessConfig, bool &$boolRunning): void
+    {
+        $arrCallback = $arrProcessConfig['callback'] ?? null;
+        if (!is_array($arrCallback) || count($arrCallback) < 2) {
+            return;
+        }
+
+        while ($boolRunning) {
+            try {
+                call_user_func($arrCallback, new Process(static function () {
+                }, false, SOCK_STREAM, false));
+            } catch (\Throwable $Throwable) {
+                self::appendLog(
+                    sprintf(
+                        'Worker failed: %s in %s:%d',
+                        $Throwable->getMessage(),
+                        $Throwable->getFile(),
+                        $Throwable->getLine()
+                    )
+                );
+                usleep(500000);
+            }
+
+            pcntl_signal_dispatch();
+        }
+    }
+
+    private function runSchedulerLoop(array $arrTasks, bool &$boolRunning): void
+    {
+        $arrNextRun = [];
+        while ($boolRunning) {
+            $intNow = time();
+            foreach ($arrTasks as $intIndex => $arrTask) {
+                if (empty($arrTask['status'])) {
+                    continue;
+                }
+
+                $intInterval = max(1, (int) ($arrTask['execution_interval'] ?? 1));
+                $intDueTime = $arrNextRun[$intIndex] ?? 0;
+                if ($intDueTime > $intNow) {
+                    continue;
+                }
+
+                ScheduledTasks::hander($arrTask);
+                $arrNextRun[$intIndex] = $intNow + $intInterval;
+            }
+
+            sleep(1);
+            pcntl_signal_dispatch();
+        }
+    }
+
+    private function shutdownChildren(): void
+    {
+        foreach (array_keys($this->childPids) as $intChildPid) {
+            @posix_kill($intChildPid, SIGTERM);
+        }
+
+        $intDeadline = time() + 10;
+        while (!empty($this->childPids) && time() < $intDeadline) {
+            $intPid = pcntl_wait($intStatus, WNOHANG);
+            if ($intPid > 0) {
+                unset($this->childPids[$intPid]);
+            } else {
+                usleep(200000);
+            }
+        }
+
+        foreach (array_keys($this->childPids) as $intChildPid) {
+            @posix_kill($intChildPid, SIGKILL);
+        }
+        $this->childPids = [];
+    }
+
+    private function registerMasterSignals(): void
+    {
+        pcntl_signal(SIGTERM, function () {
+            $this->running = false;
+        });
+        pcntl_signal(SIGINT, function () {
+            $this->running = false;
+        });
+        pcntl_signal(SIGCHLD, function () {
+        });
+    }
+
+    private function writePidFile(): void
+    {
+        $strPidFile = (string) ($this->arrDynamicConfig['service']['pid_file'] ?? '');
+        if ($strPidFile === '') {
+            return;
+        }
+
+        $strDir = dirname($strPidFile);
+        if (!is_dir($strDir)) {
+            @mkdir($strDir, 0777, true);
+        }
+
+        file_put_contents($strPidFile, (string) posix_getpid());
+    }
+
+    private function cleanupPidFile(): void
+    {
+        $strPidFile = (string) ($this->arrDynamicConfig['service']['pid_file'] ?? '');
+        if ($strPidFile !== '' && is_file($strPidFile)) {
+            @unlink($strPidFile);
+        }
+    }
+
+    private function isProcessAlive(int $intPid): bool
+    {
+        return $intPid > 0 && @posix_kill($intPid, 0);
+    }
+}