同步

@@ -15,13 +15,14 @@ function start(){
     $pm = new Process\Manager();
     $pm = new Process\Manager();
 
 
     // 启动业务进程
     // 启动业务进程
-    $pm->addBatch(20,function (Process\Pool $pool, int $worker_id){
+    $pm->addBatch(10,function (Process\Pool $pool, int $worker_id){
 
 
         swoole_set_process_name('php-email-sync-list-'.$worker_id);
         swoole_set_process_name('php-email-sync-list-'.$worker_id);
 
 
         include_once __DIR__."/../vendor/autoload.php";
         include_once __DIR__."/../vendor/autoload.php";
         _echo("业务进程（{$worker_id}）启动成功");
         _echo("业务进程（{$worker_id}）启动成功");
 
 
+        $goNum = 0;
         // 循环阻塞
         // 循环阻塞
         while (true){
         while (true){
 
 
@@ -32,10 +33,30 @@ function start(){
                 // 占用当前的id,占用2小时
                 // 占用当前的id,占用2小时
                 if(redis()->add('just_sync_'.$id,time(),600)){
                 if(redis()->add('just_sync_'.$id,time(),600)){
                     // 启动一个协程
                     // 启动一个协程
-                    go(function () use ($id){
-                        echo file_get_contents('http://mail-serve.hagro.cn/v3/sync?id='.$id);
+                    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,60);
+                            // 写入日志
+                            \Lib\Log::getInstance()->write();
                         });
                         });
 
 
+                    });
+                    while ($goNum == 0){
+                        co::sleep(0.5);
+                        break;
+                    }
                 }
                 }
             }else{
             }else{
                 co::sleep(1);
                 co::sleep(1);