优化同步

@@ -14,6 +14,7 @@ class SyncToEsCmd {
 
 
     public $isStop = false;
     public $isStop = false;
 
 
+    static $num = 0;
 
 
     public function handler(){
     public function handler(){
 
 
@@ -27,9 +28,6 @@ class SyncToEsCmd {
 //        pcntl_signal(SIGHUP, $handler);
 //        pcntl_signal(SIGHUP, $handler);
 
 
 
 
-        $es = es();
-        $db = db();
-
         while (1){
         while (1){
 
 
             // 检查是否接收到信号
             // 检查是否接收到信号
@@ -41,26 +39,45 @@ class SyncToEsCmd {
             }
             }
 
 
             $id = redis()->lPop('sync_to_es');
             $id = redis()->lPop('sync_to_es');
-            $code = 500;
             if($id){
             if($id){
-                $data = $db->first(\Model\listsSql::first('`id` = '.$id));
-                if($data){
-                    // 设置 进程 是否在运行
-                    $data['is_auto']=$db->count('select count(*) from `lists_auto` where `list_id` = '.$data['id']) ? 1 : 0;
+                while(self::$num>=50){
+                    co::sleep(0.5);
+                }
+                go(function () use ($id){
+                    static::$num++;
+                    $db = db();
+                    $code = 500;
 
 
-                    $data = $this->getEsData($data);
-                    $doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
+                    $data = $db->first(\Model\listsSql::first('`id` = '.$id));
+                    if($data){
+                        // 设置 进程 是否在运行
+                        $data['is_auto']=$db->count('select count(*) from `lists_auto` where `list_id` = '.$data['id']) ? 1 : 0;
 
 
-                    $code = $es->save($doc_id,$data);
-                }
-            }
+                        $data = $this->getEsData($data);
+                        $doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
+
+                        $code = es()->save($doc_id,$data);
+                    }
+
+                    if($code!==200){
+                        @file_put_contents(LOG_PATH.'/sync_es_fail.log',$id."\n",FILE_APPEND);
+                        _echo('同步es: '.$doc_id.'===>'.$code);
+                    }
 
 
-            if($code!==200){
-                @file_put_contents(LOG_PATH.'/sync_es_fail.log',$id."\n",FILE_APPEND);
-                _echo('同步es: '.$doc_id.'===>'.$code);
+                    $db = null;
+
+                    co::defer(function (){
+                        static::$num--;
+
+                        db()->close();
+
+                        \Lib\Log::getInstance()->write();
+                    });
+                });
+            }else{
+                co::sleep(1);
             }
             }
 
 
-            sleep(1);
         }
         }
 
 
     }
     }
@@ -91,7 +108,12 @@ class SyncToEsCmd {
 
 
 }
 }
 
 
-(new SyncToEsCmd())->handler();
+\Lib\DbPool::$clientNumber = 60;
+
+\Co\run(function (){
+    (new SyncToEsCmd())->handler();
+});
+
 
 
 return 1;
 return 1;