1

app/admin/controller/Goods.php

@@ -197,8 +197,10 @@ class Goods extends Controller
                 'update_time' => time(),
             ];
 
-            
             GoodsSku::whereIn('goods_id', array_column($list->toArray(), "goods_id"))->update($goods_sku_data);
+
+            //同步到批发商城
+            
             return $this->renderSuccess('更新成功');
         }
         return $this->renderError($model->getError() ?: '更新失败');

app/admin/controller/Store.php

@@ -18,7 +18,8 @@ use app\store\model\GoodsSku;
 use app\store\model\GoodsImage;
 use app\store\model\Category;
 use app\store\model\GoodsSpecRel;
-
+use app\job\controller\goods\GoodsRealDelete as GoodsRealDeleteJob;
+ 
 /**
  * 商城管理
  * Class Store
@@ -401,6 +402,19 @@ class Store extends Controller
                 break;
             }
             $goods_ids = array_column($goods_list->toArray(), "goods_id");
+            // 分批每次导入20条
+            $limit = 200;
+            // 根据商品总数量计算需要的队列任务数量
+            $jobCount = \count($goods_ids) / $limit;
+            // 逐次发布队列任务
+            for ($i = 0; $i < $jobCount; $i++) {
+                $data = array_slice($goods_ids, $i * $limit, $limit);
+                GoodsRealDeleteJob::dispatch([
+                        'list' => $data,
+                    ]);
+                
+            }
+
             //删除商品sku
             GoodsSku::whereIn('goods_id', $goods_ids)->delete();
             //删除商品图片

app/api/controller/Goods.php

@@ -42,16 +42,16 @@ class Goods extends Controller
         $param['channels'] = $this->storeInfo['open_channel'] ? array_merge(['zy'], explode(",", $this->storeInfo['open_channel'])) : [];
         //分类利润-利润率
         $param['fliter_condition'] = $this->storeInfo['fliter_condition'];
-        // if (isset($param['keyword']) && $param['keyword']) {
-        //     $page = $this->request->param('page', 1);
-        //     $pageSize = $this->request->param('pageSize', 10);
-        //     $param['status'] = 10;
-        //     $param['store_id'] = $this->storeId;
-        //     $goodsService = new GoodsCateEs();
-        //     $list = $goodsService->list($param, $page, $pageSize);
-        // } else {
+        if (isset($param['keyword']) && $param['keyword']) {
+            $page = $this->request->param('page', 1);
+            $pageSize = $this->request->param('pageSize', 10);
+            $param['status'] = 10;
+            $param['store_id'] = $this->storeId;
+            $goodsService = new GoodsCateEs();
+            $list = $goodsService->list($param, $page, $pageSize);
+        } else {
             $list = $model->getList($param, 10);
-        //}
+        }
         
         return $this->renderSuccess(compact('list'));
     }

app/command/SyncGoodsToEs.php

@@ -26,23 +26,17 @@ class SyncGoodsToEs extends Command
         $type = $input->getArgument("type");
         $goods_id = $input->getArgument("goods_id");
         $store_id = $input->getArgument("store_id");
-        
+        $goodsCateService = new GoodsCateEs();
+        // $goodsCateService->batchDelete([345865,1040929]);
+        // exit();
         if ($type == "goods") {
-            $goodsCateService = new GoodsCateEs();
-            //$goodsService = new GoodsEs();
-            //$goodsService->createGoodsIndex();
             $goodsCateService->batchCreateGoods($goods_id, $store_id);
         } elseif ($type == "goods_cate"){
-            $goodsCateService = new GoodsCateEs();
+            
             //$goodsCateService->createGoodsCateIndex();
             $goodsCateService->batchCreateGoodsCategory($goods_id, $store_id);
         }
         
-        // try {
-        //     $goodsService->batchCreateData();
-        // }catch (\Exception $e) {
-        //     $output->writeln($e->getMessage());
-        // }
         
     }
 

app/command/SyncGoodsToPerMinute.php

@@ -0,0 +1,52 @@
+<?php
+
+namespace app\command;
+
+use app\common\service\GoodsCateEs;
+use app\common\service\GoodsEs;
+use think\console\Command;
+use think\console\Output;
+use think\console\Input;
+use app\store\model\Goods as GoodsModel;
+use think\facade\Db;
+
+class SyncGoodsToPerMinute extends Command
+{
+
+    protected function configure()
+    {
+        // 指令配置
+        $this->setName('SyncGoodsToPerMinute')->setDescription('同步商品数据到ES');
+        $this->addArgument("goods_id");
+        $this->addArgument("div");
+        $this->addArgument("mod");
+        $this->addArgument("minute");
+    }
+
+
+    protected function execute(Input $input, Output $output)
+    {
+        ini_set('memory_limit', '1024M');
+        $goods_id = $input->getArgument("goods_id");
+        $div = $input->getArgument("div");
+        $mod = $input->getArgument("mod");
+        $minute = $input->getArgument("minute");
+
+        $min_time = time() - $minute * 60;
+        $limit = 100;
+        while (TRUE) {
+            $sql = "SELECT * FROM `yoshop_goods` WHERE goods_id > " . $goods_id . " AND goods_id  % ". $div . " = ". $mod ." AND update_time > ".$min_time." order by update_time asc LIMIT ".$limit;
+            $list = Db::query($sql);
+            if (!$list) {
+                return true;
+            }
+            $obj = new GoodsCateEs();
+            $res = $obj->addOrUpdateGoodsAndGoodsCate($list);
+            var_dump($res);
+
+
+            $goods_id = end($list)['goods_id'];
+        }
+    }
+
+}

app/common/library/elasticsearch/Client.php

@@ -171,7 +171,13 @@ class Client
         return $this->client->deleteByQuery($params);
     }
 
-
+    public function batchDelete($bulkBody = []){
+        $params = [
+            'index' => $this->index,
+            'body' => $bulkBody,
+        ];
+        return $this->client->bulk($params);
+    }
     public function existsIndex(String $index_name) {
         $params = [
             'index' => $index_name

app/common/service/GoodsCateEs.php

@@ -109,7 +109,7 @@ class GoodsCateEs
         ]);
         //var_dump($params);
         // 排序规则
-        $sort = [];
+        $sort[] = ["_score" => 'desc'];
         if ($params['sortType'] === 'all') {
             $sort[] = ['sort' => 'asc'];
         } elseif ($params['sortType'] === 'sales') {
@@ -210,7 +210,7 @@ class GoodsCateEs
             ],
 
         ];
-        return $this->esService->addDoc('category_'.$id, $doc)->asArray();
+        return $this->esService->addChildDoc('category_'.$id, $parent_id, $doc)->asArray();
     }
 
 
@@ -331,24 +331,10 @@ class GoodsCateEs
                 echo "没有商品分类了".PHP_EOL;
                 return false;
             }
-            foreach ($goods_cate_list as $value) {
-                $parent_id = 'goods_'.$value['goods_id'];
-                $doc = [
-                    'id' => $value['id'],
-                    'category_id' => $value['category_id'],
-                    'cate_store_id' => $value['store_id'],
-                    //'cate_create_time' => strtotime($value['create_time']),
-                    'cate_create_time' => $value['create_time'],
-                    'goods_category_field' => [
-                        'name' => 'category',
-                        'parent' => $parent_id
-                    ],
-
-                ];
-                //            dd($doc);
-                $res = $this->esService->addChildDoc('category_'.$value['id'], $parent_id, $doc)->asArray();
+            foreach ($goods_cate_list as $goods_cate) {
+                $res = $this->createCateData($goods_cate['id'], $goods_cate);
                 if($res['result'] == 'created' || $res['result'] == 'updated'){
-                    echo('商品分类同步成功'.$value['id'].'条数据').PHP_EOL;
+                    echo('商品分类同步成功'.$goods_cate['id'].'条数据').PHP_EOL;
                 }
             }
             $goods_cate_id = end($goods_cate_list)['id'];
@@ -369,7 +355,6 @@ class GoodsCateEs
             echo $goods_id.PHP_EOL;
             $goods_list = Goods::where('goods_id', '>', $goods_id)
                 ->where($where)
-                //->field("goods_id,goods_name,goods_no,content,cost_price_min,goods_price_min,sort,sales_actual,profit,profit_rate,channel,store_id,is_delete,status,create_time")
                 ->order('goods_id asc')
                 ->limit(1000)
                 ->select()
@@ -379,15 +364,10 @@ class GoodsCateEs
                 echo "没有商品了".PHP_EOL;
                 return false;
             }  
-            foreach ($goods_list as $value) {
-               
-                $value['goods_category_field'] = [
-                    'name' => 'goods'
-                ];
-                $res = $this->esService->addDoc('goods_'.$value['goods_id'], $value)->asArray();
-                //var_dump($res);
+            foreach ($goods_list as $goods) {
+                $res = $this->createData($goods['goods_id'], $goods);
                 if($res['result'] == 'created' || $res['result'] == 'updated'){
-                    echo('商品同步成功'.$value['goods_id'].'条数据').PHP_EOL;
+                    echo('商品同步成功'.$goods['goods_id'].'条数据').PHP_EOL;
                 }
             }
             $goods_id = end($goods_list)['goods_id'];
@@ -403,16 +383,45 @@ class GoodsCateEs
         if (!$goods_list) {
             return true;
         }
-        foreach ($goods_list as $goods) {
+        $goods_cate_list = GoodsCategoryRel::whereIn('goods_id', array_column($goods_list, "goods_id"))->select()->toArray();
+        $arr = [];
+        foreach ($goods_cate_list as $key => $value) {
+            $arr[$value['goods_id']][] = $value;
+        }
+        foreach ($goods_list as &$goods) {
+            $goods['create_time'] = date("Y-m-d H:i:s", $goods['create_time']);
+            $goods['update_time'] = date("Y-m-d H:i:s", $goods['update_time']);
             $this->createData($goods['goods_id'], $goods);
-            $goods_cate_list = GoodsCategoryRel::where('goods_id', $goods['goods_id'])->select()->toArray();
-            foreach ($goods_cate_list as $goods_cate) {
+            $goods_cate_items = $arr[$goods['goods_id']] ?? [];
+            if (!$goods_cate_items) {
+                continue;
+            }
+            foreach ($goods_cate_items as $goods_cate) {
                 $this->createCateData($goods_cate['id'], $goods_cate);
             }
         }
         return true;
         
     }
+    /**
+     * 批量删除父子关系的数据
+     * [batchDelete description]
+     * @param  [type] $parentIds [description]
+     * @return [type]            [description]
+     */
+    public function batchDelete($parentIds){
+        $bulkBody = [];
+        foreach ($parentIds as $parentId) {
+            $bulkBody[] = [
+                'delete' => [
+                    '_index' => $this->index_name,
+                    //'_type' => '_doc', // 在 Elasticsearch 7.x 之前的版本中，类型字段是必需的
+                    '_id' => 'goods_'.$parentId
+                ]
+            ];
+        }
+        return $this->esService->batchDelete($bulkBody);
+    }
     /**
      * 删除商品
      */

app/job/controller/goods/GoodsRealDelete.php

@@ -0,0 +1,40 @@
+<?php
+declare (strict_types=1);
+
+namespace app\job\controller\goods;
+
+use app\job\service\goods\GoodsRealDelete as GoodsRealDeleteService;
+use cores\BaseJob;
+use cores\traits\QueueTrait;
+use cores\exception\BaseException;
+use think\db\exception\DataNotFoundException;
+use think\db\exception\DbException;
+use think\db\exception\ModelNotFoundException;
+
+/**
+ * 队列任务：商品批量导入
+ * Class Import
+ * @package app\job\controller
+ */
+class GoodsRealDelete extends BaseJob
+{
+    use QueueTrait;
+
+    /**
+     * 消费队列任务：商品导入
+     * @param array $data 参数 [index队列顺序；totalCount商品总数量；list商品列表；storeId商城ID]
+     * @return bool 返回结果
+     * @throws BaseException
+     * @throws DataNotFoundException
+     * @throws DbException
+     * @throws ModelNotFoundException
+     */
+    public function handle(array $data): bool
+    {
+        $time = date('H:i:s');
+        echo "\n  ---- GoodsRealDelete ----  {$time} ----   \n";
+
+        $service = new GoodsRealDeleteService;
+        return $service->batch($data['list']);
+    }
+}

app/job/service/goods/GoodsRealDelete.php

@@ -0,0 +1,41 @@
+<?php
+declare (strict_types=1);
+
+namespace app\job\service\goods;
+
+use app\common\library\helper;
+use app\common\service\BaseService;
+use app\common\model\Goods as GoodsModel;
+use app\common\model\GoodsSku;
+use app\store\model\GoodsCategoryRel;
+use cores\exception\BaseException;
+use think\facade\Log;
+use app\common\service\GoodsCateEs;
+
+/**
+ * 批量同步到es
+ * Class Import
+ * @package app\job\service\goods
+ */
+class GoodsRealDelete extends BaseService
+{
+    /**
+     * 批量同步到es
+     * @param array $list
+     * @param int $recordId
+     * @param int $storeId
+     * @return bool
+     * @throws BaseException
+     * @throws \think\db\exception\DataNotFoundException
+     * @throws \think\db\exception\DbException
+     * @throws \think\db\exception\ModelNotFoundException
+     */
+    public function batch(array $goodsIds): bool
+    {
+        $obj = new GoodsCateEs();
+        $obj->batchDelete($goodsIds);
+        return true;
+    }
+
+   
+}

config/console.php

@@ -17,5 +17,6 @@ return [
         'ProfitSharingResult' => 'app\command\ProfitSharingResult',
         'SyncCategory' => 'app\command\SyncCategory',
         'HomeLocation' => 'app\command\HomeLocation',
+        'SyncGoodsToPerMinute' => 'app\command\SyncGoodsToPerMinute',
     ],
 ];