debug

code/app/task/core/Core.php

@@ -0,0 +1,129 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\task\core;
+
+class Core
+{
+    static $HttpService;
+
+    static $arrDynamicConfig = [];
+
+    public function __init($arrArgvs = [])
+    {
+        $this->setConfigDaemonize($arrArgvs[0]['daemonize'] ?? 'false');
+    }
+
+    private function setConfigDaemonize($strMode)
+    {
+        self::$arrDynamicConfig['daemonize'] = (bool)$strMode;
+    }
+
+    private function getPid()
+    {
+        return file_exists(config('task.service.pid_file')) ? file_get_contents(config('task.service.pid_file')) : NULL;
+    }
+
+    public function start($arrArgvs = [])
+    {
+        $this->__init($arrArgvs);
+
+        $intPid = (int)$this->getPid();
+
+        if ($intPid > 0 && \Swoole\Process::kill($intPid, SIG_DFL)) {
+            echo "  Service is \e[0;32mRunning\e[0m !  " . PHP_EOL;
+            exit;
+        }
+        echo "  Execute the \e[0;32m start\e[0m command ......   " . PHP_EOL;
+
+        $this->createService();
+
+        $this->addProcessList();
+       
+        $this->addScheduledTasks();
+     
+        self::$HttpService->getServer()->start();
+    
+    }
+
+    public function stop($arrArgvs = [])
+    {
+        $this->__init($arrArgvs);
+
+        $intPid = (int)$this->getPid();
+
+        if ($intPid <= 0) {
+            echo "  Service is \e[0;31mStop\e[0m !  " . PHP_EOL;
+        } else {
+            echo "  Execute the \e[0;32m stop\e[0m command ......   " . PHP_EOL;
+            \Swoole\Process::kill($intPid, SIGTERM);
+            echo "  Service is \e[0;31mStop\e[0m !  " . PHP_EOL;
+        }
+    }
+
+    public function status($arrArgvs = [])
+    {
+        $this->__init($arrArgvs);
+
+        $intPid = (int)$this->getPid();
+
+        if ($intPid <= 0) {
+            echo "  Service is \e[0;31mStop\e[0m !  " . PHP_EOL;
+        } else if (\Swoole\Process::kill($intPid, SIG_DFL)) {
+            echo "  Service is \e[0;32mRunning\e[0m !  " . PHP_EOL;
+        } else {
+            echo "  Service is \e[0;31mStop\e[0m !  " . PHP_EOL;
+        }
+    }
+
+    public function restart($arrArgvs = [])
+    {
+        $this->__init($arrArgvs);
+
+        $intPid = (int)$this->getPid();
+
+        if ($intPid <= 0) {
+            echo "  \e[0;31m Get Pid Error\e[0m !" . PHP_EOL;
+            echo "  Execute the \e[0;32m start\e[0m command ......   " . PHP_EOL;
+            $this->start();
+        } else {
+            echo "  Execute the \e[0;32m restart\e[0m command ......   " . PHP_EOL;
+            \Swoole\Process::kill($intPid, SIGUSR1);
+        }
+    }
+
+    private function addScheduledTasks()
+    {
+        if (config('task.scheduled_tasks_pool')) {
+            foreach (config('task.scheduled_tasks_pool') as $arrProcessTask) {
+                if ($arrProcessTask['status']) {
+                    $Process = new \Swoole\Process(function () use ($arrProcessTask) {
+                        call_user_func([\app\task\core\ScheduledTasks::class, 'hander'], $arrProcessTask);
+                    });
+                    self::$HttpService->getServer()->addProcess($Process);
+                }
+            }
+        }
+    }
+
+    private function addProcessList()
+    {
+        if (config('task.process_pool')) {
+            foreach (config('task.process_pool') as $v) {
+                for ($i = 0; $i < $v['Num']; $i++) {
+                    $Process = new \Swoole\Process($v['callback']);
+
+                    self::$HttpService->getServer()->addProcess($Process);
+                }
+            }
+        }
+    }
+
+    private function createService()
+    {
+        $arrHttpServiceConfig = array_merge(config('task.service'), self::$arrDynamicConfig);
+
+        self::$HttpService = Service::instance($arrHttpServiceConfig);
+    }
+}

code/app/task/core/Listen.php

@@ -0,0 +1,90 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\task\core;
+
+use app\task\logic\SystemLogic;
+
+class Listen
+{
+    /**
+     * 监听队列1代理
+     *
+     * @param \Swoole\Process $Process
+     * @return void
+     */
+    static public function innerConsume(\Swoole\Process $Process)
+    {
+        return self::innerConsumeCore($Process, 1);
+    }
+
+
+    /**
+     * 监听队列2代理
+     *
+     * @param \Swoole\Process $Process
+     * @return void
+     */
+    static public function innerConsume02(\Swoole\Process $Process)
+    {
+        return self::innerConsumeCore($Process, 2);
+    }
+
+    /**
+     * 监听队列
+     *
+     * @param \Swoole\Process $Process
+     * @return void
+     */
+    static private function innerConsumeCore(\Swoole\Process $Process, int $intIndex = 1)
+    {
+        Service::$Process = $Process;
+
+        SystemLogic::destructConnectSource();
+
+        $arrConfig = config('task.queue');
+
+        $intLimit =  $arrConfig[$intIndex]['exec_num'];
+
+        $TaskCore = new TaskCore($intIndex);
+
+        return self::excuteTask($TaskCore, $intLimit);
+    }
+
+    /**
+     * 队列任务消费
+     *
+     * @param \Swoole\Process $Process
+     * @return void
+     */
+    static private function excuteTask(TaskCore $TaskCore, $intLimit)
+    {
+        while ($intLimit > 0) {
+            $intLimit--;
+            try {
+                $strData = $TaskCore->get();
+                if (empty($strData)) {
+                    sleep(1);
+                    continue;
+                }
+                $arrData = json_decode($strData, true);
+                if (!is_array($arrData)) {
+                    echo '监听Redis队列读取到异常无法解析的数据[' . $strData . ']' . PHP_EOL;
+                    continue;
+                }
+                $Class = new $arrData['callback'][0];
+                $Class->{$arrData['callback'][1]}($arrData['data']);
+                unset($Class);
+            } catch (\Throwable $t) {
+                $strData = $strData ?? '';
+                echo date('Y-m-d H:i:s') . ':出现致命错误需要处理' . PHP_EOL .
+                    '           队列数据:' . $strData . PHP_EOL .
+                    '           文件:' . $t->getFile() . PHP_EOL .
+                    '           行数:' . $t->getLine() . PHP_EOL .
+                    '           错误描述:' . $t->getMessage() . PHP_EOL .
+                    '           堆栈跟踪:' . $t->getTraceAsString() . PHP_EOL;
+            }
+        }
+    }
+}

code/app/task/core/Rout.php

@@ -0,0 +1,31 @@
+<?php 
+declare (strict_types = 1);
+
+namespace app\task\core;
+
+class Rout
+{
+    static public function httpDispense(\Swoole\Http\Request $Request, \Swoole\Http\Response $Response)
+    {
+
+      //  print_r($Request->getData());
+     //   print_r($Request->server);
+
+        # 简单的HTTP服务
+        $Response->status(999,'Hei Guys ~');
+        $Response->header("Content-Type", "text/html; charset=utf-8");
+        $Response->end("<h1>Hello reptile~. #".rand(1000, 9999)."</h1>");
+    }
+
+    static public function tcpDispense(\Swoole\Server $server, $fd, $reactor_id, $mixedData)
+    {
+        # 简单的HTTP服务
+        print_r($mixedData);
+    }
+
+    static public function test()
+    {
+       var_dump(date('Y-m-d H:i:s'));
+       return;
+    }
+}

code/app/task/core/ScheduledTasks.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\task\core;
+
+use app\task\logic\SystemLogic;
+
+class ScheduledTasks
+{
+    static public function hander($arrProcessTask = [])
+    {
+        SystemLogic::destructConnectSource();
+        //  while(true){
+        call_user_func($arrProcessTask['callback'], $arrProcessTask['param']);
+
+        sleep($arrProcessTask['execution_interval']);
+        // }
+
+        // \Swoole\Timer::tick($arrProcessTask['execution_interval']*1000, $arrProcessTask['callback'], $arrProcessTask['param']);
+
+        // \Swoole\Event::wait();
+    }
+}

code/app/task/core/Service.php

@@ -0,0 +1,87 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\task\core;
+
+class Service
+{
+    private $arrConfig;
+
+    private static $obj;
+
+    private $Server;
+
+    static $Process = NULL;
+
+    static function instance($arrConfig)
+    {
+        if (self::$obj == null) {
+            self::$obj = new self($arrConfig);
+        }
+        return self::$obj;
+    }
+
+    public function __construct($arrConfig)
+    {
+        $this->arrConfig = $arrConfig;
+        $this->createServer();
+    }
+
+    private function serHttpConfig()
+    {
+        $this->Server->set($this->filterConfig());
+        $this->Server->on('request', [\app\task\core\Rout::class, 'httpDispense']);
+    }
+
+    private function filterConfig()
+    {
+        $arrConfig = $this->arrConfig;
+        unset($arrConfig['host']);
+        unset($arrConfig['port']);
+        unset($arrConfig['mode']);
+        unset($arrConfig['sockType']);
+        unset($arrConfig['server_type']);
+        return $arrConfig;
+    }
+
+    private function serTcpConfig()
+    {
+        $this->Server->set($this->filterConfig());
+        $this->Server->on('receive', [\app\task\core\Rout::class, 'tcpDispense']);
+    }
+
+    private function createServer()
+    {
+        switch ($this->arrConfig['server_type']) {
+            case 'HTTP':
+                $this->Server = new \Swoole\Http\Server($this->arrConfig['host'], $this->arrConfig['port']);
+                $this->serHttpConfig();
+                break;
+            default:
+                $this->Server = new \Swoole\Server($this->arrConfig['host'], $this->arrConfig['port'], $this->arrConfig['mode'], $this->arrConfig['sockType']);
+                $this->serTcpConfig();
+                break;
+        }
+
+        $this->Server->on('task', [$this, 'onTask']);
+
+        $this->Server->on('finish', [$this, 'onFinish']);
+    }
+
+    public function getServer()
+    {
+        return $this->Server;
+    }
+
+    public function onTask($Http, $task_id, $from_id, $arrData)
+    {
+        $mixedResult = call_user_func_array($arrData[0], $arrData[1]);
+
+        $Http->finish($mixedResult);
+    }
+
+    public function onFinish($Http, $task_id, $mixedData)
+    {
+    }
+}

code/app/task/core/TaskCore.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\task\core;
+
+use think\facade\Cache;
+
+class TaskCore
+{
+    public $strQueueName = '';
+
+    public function __construct($intIndex = 1, $strQueueName = '')
+    {
+        if ($strQueueName == '') {
+            $arrConfig = config('task.queue');
+            $strQueueName = $arrConfig[$intIndex]['name'];
+        }
+
+        $this->strQueueName = $strQueueName;
+    }
+
+    public function get()
+    {
+        return Cache::store('redis')->lpop($this->strQueueName);
+    }
+
+    public function set($arrData)
+    {
+        $strData = json_encode($arrData);
+        // 打印调试信息
+        //\think\facade\Log::info('TaskCore set: '.$strData );
+        return Cache::store('redis')->rpush($this->strQueueName, $strData);
+    }
+}