x

@@ -4,43 +4,82 @@
 
 
 use Swoole\Process;
 use Swoole\Process;
 
 
-include_once __DIR__."/../vendor/autoload.php";
-
-//exec("ps -aux|grep -E \"mail-serve-sync-list\" -c",$exec);
-//if(intval($exec[0]) > 200){
-//    exit;
-//}
-
-swoole_set_process_name("mail-serve-sync-list");
-
-while (1){
-    $id = redis()->lPop('sync_email_lists');
-    if($id && is_numeric($id)){
-        // 占用当前的id,占用2小时
-        if(redis()->add('just_sync_'.$id,time(),600)){
-            // 启动一个协程
-
-            try{
-                // 开始同步
-                (new \Service\SyncMail($id))->sync();
-            }catch (Throwable $e){
-                logs('sync : '.$e->getMessage());
+
+
+function start(){
+
+// 删除停止运行的值
+//    redis()->delete(SYNC_RUNNING_REDIS_KEY,'email_sync_stop_num');
+
+    // 进程管理器
+    $pm = new Process\Manager();
+
+    // 启动业务进程
+    $pm->addBatch(10,function (Process\Pool $pool, int $worker_id){
+
+        swoole_set_process_name('php-email-sync-list-'.$worker_id);
+
+        include_once __DIR__."/../vendor/autoload.php";
+        _echo("业务进程（{$worker_id}）启动成功");
+
+        $goNum = 0;
+        // 循环阻塞
+        while (true){
+
+            // 需要同步的id
+            $id = redis()->lPop('sync_email_lists');
+
+            if($id && is_numeric($id)){
+                // 占用当前的id,占用2小时
+                if(redis()->add('just_sync_'.$id,time(),600)){
+                    // 启动一个协程
+                    go(function () use ($id,&$goNum){
+                        $goNum++;
+                        try{
+                            // 开始同步
+                            (new \Service\SyncMail($id))->sync();
+                        }catch (Throwable $e){
+                            logs('sync : '.$e->getMessage());
+                        }
+
+
+                        // 协程完成后执行的函数
+                        co::defer(function () use ($id,&$goNum){
+                            $goNum--;
+                            // 30秒后 消除占用
+                            redis()->expire('just_sync_'.$id,300);
+                            // 写入日志
+                            \Lib\Log::getInstance()->write();
+                        });
+
+                    });
+                }
             }
             }
 
 
+            //每次都暂停1秒，防止同一时间启动太多的任务
+            co::sleep(0.5);
 
 
-            // 30秒后 消除占用
-            redis()->expire('just_sync_'.$id,300);
+            while ($goNum > 50){
+                co::sleep(1);
+                break;
+            }
 
 
-            break;
         }
         }
-    }
 
 
-    usleep(1000);
+    },true);
+
+
+    // 启动管理器
+    $pm->start();
+
 }
 }
 
 
 
 
 
 
 
 
+start();
+
+