同步到es 改成批量

邓超
cmd/sync_to_es_v2.php
lib/Es/BulkData.php
--- a/cmd/sync_to_es_v2.php
查看文件 @4449166
+++ b/cmd/sync_to_es_v2.php
查看文件 @4449166
@@ -89,16 +89,50 @@ class SyncToEsCmd {
      */
     public $es;
+    /**
+     * @var \Lib\Es\BulkData
+     */
+    protected $bulkData;
+
     public function handler(){
         $this->es = es('email_lists_copy'); // 第二个库 新
-        while (!$this->isStop()){
+        $this->bulkData = new \Lib\Es\BulkData();
+
+        $ids = [];
+        while (!$this->isStop()){
+            $nowSubmit = false;
             $id = redis()->lPop('sync_to_es');
             if($id){
+                $ids[] = $id;
+                $this->bulkData($id);
+            }else{
+                $nowSubmit = true;
+                // 没有数据时暂停1秒
+                sleep(1);
+            }
+            if(!$this->toDataEs($nowSubmit)){
+                $this->log($ids);
+                $ids = [];
+            }
+        }
+
+        if(!$this->toDataEs(true)){
+            $this->log($ids);
+        }
+    }
+
+    /**
+     * 批量处理数据并存储到ES
+     *
+     * @param string $id 数据ID，如果包含点号则只取点号前的部分
+     * @return void
+     */
+    public function bulkData($id){
         $is_check_body = false;
         if(str_contains($id, '.')){
             $id = explode('.',$id)[0];
@@ -108,9 +142,8 @@ class SyncToEsCmd {
         $data = $this->getDataByEs($id,$is_check_body);
         if($data){
             list($doc_id,$data) = $data;
-                    $bulkData = new \Lib\Es\BulkData();
             // 主库
-                    $bulkData->add('email_lists_copy',$doc_id,$data);
+            $this->bulkData->add('email_lists_copy',$doc_id,$data);
             // 个人邮箱的情况
             if(!$data['postid']){
                 $postids = $this->getPostids($data['email_id']);
@@ -119,27 +152,20 @@ class SyncToEsCmd {
                         $data['postid'] = $postid;
                         $data['source'] = 1;
                         // 分库 个人邮箱
-                                $bulkData ->add('email_lists_branch_'.$postid,$doc_id,$data);
+                        $this->bulkData ->add('email_lists_branch_'.$postid,$doc_id,$data);
                     }
                 }else{
                     // 分库 其他 非fob数据源
-                            $bulkData->add('email_lists_branch_0',$doc_id,$data);
+                    $this->bulkData->add('email_lists_branch_0',$doc_id,$data);
                 }
             }else{
                 // 分库
-                        $bulkData->add('email_lists_branch_'.$data['postid'],$doc_id,$data);
+                $this->bulkData->add('email_lists_branch_'.$data['postid'],$doc_id,$data);
             }
-                    $this->toDataEs($id,$bulkData);
-
-                }
-            }else{
-                // 没有数据时暂停1秒
-                sleep(1);
         }
     }
-    }
     /**
      * 个人邮箱情况
@@ -173,7 +199,7 @@ class SyncToEsCmd {
                 $data = $this->db->throw()->first(\Model\listsSql::firstHot('`id` = '.$id));
             }
         }catch (Throwable $e){
-            $this->log($id);
+            $this->log([$id]);
 //            redis()->rPush('sync_to_es',$origin_id);
             _echo('sync to es '.$id.":".$e->getMessage());
             return false;
@@ -208,7 +234,7 @@ class SyncToEsCmd {
                 list($data['postid'],$data['source']) = $this->getPostid($data['email_id'],$data['udate']);
             }catch (Throwable $e){
-                $this->log($id);
+                $this->log([$id]);
 //                redis()->rPush('sync_to_es',$origin_id);
                 _echo('sync to es '.$id.":".$e->getMessage());
                 return false;
@@ -287,19 +313,35 @@ class SyncToEsCmd {
         return $this->setEsMap($index);
     }
+    /**
+     * @var int
+     */
+    protected $pervSubmitTime = 0;
     /**
      * 同步数据到es
-     * @param int $data_id
-     * @param \Lib\Es\BulkData $bulkData
+     * @param bool $nowSubmit 是否立即提交
+     * @return bool
      * @throws \Elastic\Elasticsearch\Exception\ClientResponseException
      * @throws \Elastic\Elasticsearch\Exception\ServerResponseException
      * @author：dc
-     * @time 2025/8/6 20:55
+     * @time 2025/8/7 10:29
      */
-    public function toDataEs(int $data_id, \Lib\Es\BulkData $bulkData){
+    public function toDataEs(bool $nowSubmit){
+        // 不立即提交
+        if (!$nowSubmit){
+            if($this->bulkData->total() < 10){
+                // 不足10条时 满足2秒也提交
+                if ($this->pervSubmitTime + 2 > time()){
+                    return true;
+                }
+            }
+        }
-        foreach ($bulkData->getIndexs() as $index){
+        // 上一次提交的时间
+        $this->pervSubmitTime = time();
+
+        foreach ($this->bulkData->getIndexs() as $index){
             $this->es->setIndex($index);
             // 检查数据库是否存在
             if(empty($this->checkEsIndex[$index]) && $index != 'email_lists_copy'){
@@ -313,24 +355,26 @@ class SyncToEsCmd {
             $this->checkEsIndex[$index] = 1;
         }
         // 批量提交数据的
-        $ret = $this->es->bulk($bulkData);
+        $ret = $this->es->bulk($this->bulkData);
+
         if(!empty($ret['errors'])){
-            $this->log($data_id);
-            @file_put_contents(LOG_PATH.'/sync_es_fail.log',$data_id."\n",FILE_APPEND);
-            _echo($data_id.' ===> 0');
+            @file_put_contents(LOG_PATH.'/sync_es_fail.error.log',print_r($ret['errors'],1)."\n",FILE_APPEND|LOCK_EX);
         }
-
+        // 清空
+        $this->bulkData->clear();
+        // 为空表示提交成功
+        return empty($ret['errors']);
     }
     /**
      * 记录日志
-     * @param $id
+     * @param array $ids
      * @param string $index
      * @author：dc
      * @time 2025/8/5 10:17
      */
-    public function log($id,$index=''){
-        @file_put_contents(LOG_PATH.'/sync_es_fail.log',$index.":".$id."\n",FILE_APPEND);
+    public function log(array $ids){
+        file_put_contents(LOG_PATH.'/sync_es_fail.log',implode("\n",$ids)."\n",FILE_APPEND|LOCK_EX);
     }
--- a/lib/Es/BulkData.php
查看文件 @4449166
+++ b/lib/Es/BulkData.php
查看文件 @4449166
@@ -67,7 +67,36 @@ class BulkData {
         return $params;
     }
+    /**
+     * 清除 数据
+     * @author：dc
+     * @time 2025/8/7 9:55
+     */
+    public function clear(){
+        $this->data = [];
+    }
+    /**
+     * 是否为空
+     * @return bool
+     * @author：dc
+     * @time 2025/8/7 9:54
+     */
+    public function isEmpty():bool
+    {
+        return empty($this->data);
+    }
+
+    /**
+     * 统计数量
+     * @return int
+     * @author：dc
+     * @time 2025/8/7 9:56
+     */
+    public function total():int
+    {
+        return count($this->data);
+    }
 }