go

邓超
cmd/send_job.php
lib/SwGo.php
--- a/cmd/send_job.php
查看文件 @6db3a8a
+++ b/cmd/send_job.php
查看文件 @6db3a8a
@@ -5,66 +5,14 @@ include_once "../vendor/autoload.php";
 // 这里试试不用多进程模式，用多协程模式
 \Lib\DbPool::$clientNumber = 60;
+\Lib\RedisPool::$clientNumber = 60;
 class SendJob {
-    public $cnum = 0;
-
-    private $run_timer = 0;
-
-    public function __construct()
-    {
-        $this->run_timer = time();
-    }
-
-    /**
-     * 是否停止
-     * @return bool
-     * @author：dc
-     * @time 2024/4/10 9:12
-     */
-    private function isStop(){
-        // 运行超过1天的 停止
-        if($this->run_timer < (time()-43200)){
-//            @posix_kill(getmypid(), SIGTERM);
-            return true;
-        }
-        return redis()->get('send_job_is_stop') == 'stop';
-    }
-
-    /**
-     * 休眠
-     * @param float $sleep
-     * @return bool
-     * @author：dc
-     * @time 2024/4/10 9:12
-     */
-    private function s_sleep(float $sleep):bool {
-        if($sleep > 0){
-            $t = microtime(1);
-
-            while (!$this->isStop()){
-                co::sleep(0.1);
-                if($sleep - (microtime(1)-$t) <= 0){
-                    break;
-                }
-            }
-        }
-        return true;
-    }
-
-
     public function start(){
         _echo('启动邮件群发任务 '.getmypid());
-        // 删除key
-        redis()->delete('send_job_is_stop');
-
         while (1){
-            // 是否要停止
-            if($this->isStop()){
-                break;
-            }
             $lists  =   db()->all(\Model\sendJobsSql::sendList(500));
             $lists = $lists?$lists:[];
@@ -92,20 +40,10 @@ class SendJob {
                 }
             }
                 // 休眠30秒
-            $this->s_sleep(30);
+            co::sleep(30);
         }
-
-        // 这个是等待所有协程退出
-        while (true){
-            _echo('等待协程退出...');
-            if(!$this->cnum){
-                break;
-            }
-            co::sleep(1);
-        }
-
     }
@@ -118,16 +56,16 @@ class SendJob {
      */
     public function go_($list){
         // 控制50个协程内
-        while ($this->cnum>=50){
-            co::sleep(0.5);
+        while (\Lib\SwGo::$runNumber >= 50){
+            co::sleep(1);
         }
         // 占用 id
         if(redis()->add('send_job_run_id_'.$list['id'],$list['id'],600)){
-            go(function ($data) {
+            \Lib\SwGo::start(function ($data) {
                 _echo('正在执行任务 '.$data['id']);
-                $this->cnum++; // 协程数+1
+
                 // 表单数据
                 $data['maildata'] = json_decode($data['maildata'],true);
                 // 查询邮箱
@@ -184,13 +122,8 @@ class SendJob {
                                 _echo('进入时间等待区 '.$to['email'].' 等待：'.$time);
                                 $block = false;
                                 while (true){
-                                    // 没5秒循环一次
-                                    if($this->isStop()){
-                                        $block = true;
-                                        break;
-                                    }
                                     $time-=5;
-                                    $this->s_sleep(5);
+                                    co::sleep(5);
                                     // 执行下一次了
                                     if (!$time){
                                         $block = true;
@@ -231,9 +164,7 @@ class SendJob {
                 }
-                // 协程结束后
-                co::defer(function ($id) use($data){
-                    $this->cnum--;
+            },function ($data){
                 // 验证是否完成
                 if($data['maildata']['massSuit']??0){
                     $dst = db()->first(\Model\sendJobsSql::isStatus($data['id']));
@@ -249,74 +180,22 @@ class SendJob {
                         }
                     }
                 }
-
-                    // 写入日志
-                    \Lib\Log::getInstance()->write();
-
                 // 删除占用
                 redis()->delete('send_job_run_id_'.$data['id']);
-
                 _echo('执行任务完成'.$data['id']);
-                    db()->close();
-                });
-
             },$list);
         }
     }
 }
-
-
-$ps = "ps -ef | grep \"send_job.php start\" | grep -v grep | wc -l";
-
-switch ($argv[1]??0){
-    case 'start':{
-
-        // 开启协程
-        \Co\run(function (){
-
-            $handler = function ($signal){
-                // 可以处理其他程序
-                redis()->set('send_job_is_stop','stop');
-
-                _echo('收到退出信号 '.$signal);
-            };
-
-            \Swoole\Process::signal(SIGTERM,$handler);
-            \Swoole\Process::signal(SIGINT,$handler);
+// 开启协程
+\Co\run(function (){
     (new SendJob)->start();
     _echo('进程已退出');
-            db()->close();
-        });
-
-        break;
-    }
-    case 'stop':{
-        \Co\run(function ($ps){
-            echo "正在退出程序...\n非必要请不要强制kill掉进程\n";
-
-            redis()->set('send_job_is_stop','stop');
-
-            while (true){
-
-                $num = exec($ps);
-                if(!$num){
-                    break;
-                }
-                co::sleep(0.2);
-            }
-            echo "已退出程序\n";
-        },$ps);
-
-        break;
-    }
-    default:{
-        break;
-    }
-}
+});
--- a/lib/SwGo.php
查看文件 @6db3a8a
+++ b/lib/SwGo.php
查看文件 @6db3a8a
@@ -37,9 +37,17 @@ class SwGo {
         go(function (\Closure $run,...$param){
             self::$runNumber++;
+            $end = null;
+            if(empty($param[0]) && $param[0] instanceof \Closure){
+                $end = $param[0];
+                unset($param[0]);
+                $param = array_values($param);
+            }
             $run(...$param);
+            if($end) $end(...$param);
+
             // 写入日志
             \Lib\Log::getInstance()->write();
             // 释放 mysql