update

2025-09-13 01:22:15 +08:00
parent 155e05fd6d
commit 1a4b8551a0
674 changed files with 146276 additions and 0 deletions
--- a/module/Application/src/Command/Swoole/Server/SwTaskCommand.php
+++ b/module/Application/src/Command/Swoole/Server/SwTaskCommand.php
@ -0,0 +1,84 @@
+<?php
+/**
+ *
+ * @author：llbjj
+ * @DateTime：2022/9/1 15:00
+ * @Description：
+ *
+ */
+namespace Application\Command\Swoole\Server;
+
+use Application\Command\BasicCommand;
+use Application\Service\Extension\ErrorHandler;
+use Application\Service\Extension\Laminas;
+use Swoole\Server;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class SwTaskCommand extends BasicCommand{
+    function execute(InputInterface $input, OutputInterface $output)
+    {
+        error_reporting(0);
+        $tcpSv = new Server('127.0.0.1', 9511);
+        $tcpSv->set([
+            'worker_num' => 4, // 工作进程数量
+            'task_worker_num' => 8,  // 异步任务进程数
+            'open_eof_check' => true,   //打开EOF检测
+            'package_eof'    => '|PHENOL|', //设置EOF
+            'package_max_length' => 1024 * 1024 * 5,
+            'dispatch_mode' => 3,
+            'task_ipc_mode' => 2,
+
+            // 'daemonize' => 1, //以守护进程 1或0
+            // 'open_eof_split' => true, //swoole底层实现自动分包。比较消耗cpu资源
+            // 'package_eof' => "|PHENOL|", //设置后缀，一般为"
+        ]);
+
+        //接收数据
+        $tcpSv->on('receive', function ($sv, $fd, $reactorId, $data) use($output) {
+            $sendDataArr = array_filter(explode('|PHENOL|', $data));
+            if(!empty($sendDataArr)) {
+                foreach($sendDataArr as $sendData) {
+                    $sv->task($sendData);
+                }
+            }
+        });
+
+        // 定义异步任务
+        $tcpSv->on('task', function($sv, $task_id, $src_work_id, $data) use ($output){
+            $taskData = json_decode($data, true);
+            try {
+                // 设置Token
+                Laminas::$token = $taskData['token'] ?? '';
+                // 调用方法，
+                $result = $this->LocalService()->{$taskData['svName']}->{$taskData['methodName']}($taskData['params']);
+
+                $sv->finish("{$data}");
+            }catch (\Throwable $throwable) {
+                $this->saveErrLog($throwable, $taskData);
+            }
+
+        });
+
+        //处理异步任务的结果(此回调函数在worker进程中执行)
+        $tcpSv->on('Finish', function ($serv, $task_id, $data) {
+            $taskData = json_decode($data, true);
+            // 发送完成通知
+            $msgData = [
+                'title' => "【{$taskData['domainName']}】异步任务 {$taskData['svName']}->{$taskData['methodName']}() 执行完成！",
+                'description' => "AsyncTask[{$task_id}] Finish: {$data}",
+                'url' => '#'
+            ];
+
+            $this->LocalService()->wechatWork->sendMesToUser($msgData);
+            echo "AsyncTask[{$task_id}] Finish: {$data} -> OK".PHP_EOL;
+        });
+
+        $tcpSv->start();
+        return 0;
+    }
+
+    public function saveErrLog(&$throwable, &$taskData) {
+        ErrorHandler::log2txt($throwable, '', $taskData);
+    }
+}