message_data = $message_data; } /** * Execute the job. * * @return void */ public function handle() { try { $ScrapeDataModel = new ScrapeDataModel(); $limit = isset($message_data['limit']) ? $message_data['limit'] : 10000; $page = isset($message_data['page']) ? $message_data['page'] : 1; $company_id = isset($message_data['company_id']) ? $message_data['company_id'] : 0; //品牌方公司ID $start_time = $this->message_data['start_time']; $end_time = $this->message_data['end_time']; $start_time_string = date('Y-m-d H:i:s', $start_time); $end_time_string = date('Y-m-d H:i:s', $end_time); $key_name = 'ScrapeDataProductJobs_' . $company_id . '_' . $start_time . '_' . $end_time; $product_datas = Cache::get($key_name); if (empty($product_datas)) { $where = []; $where[] = ['insert_time', '>=', $start_time_string]; $where[] = ['insert_time', '<=', $end_time_string]; $where[] = ['min_price', '>=', '0.01']; $where[] = ['number', '>=', '1']; $where[] = ['enterprise_id', '=', $company_id]; $product_data_info = $ScrapeDataModel->select('*', DB::raw('ROUND(min_price / NULLIF(number, 0), 2) as unit_price')) ->where($where)->orderbyDesc('id')->paginate($limit, ['*'], 'page', $page)->toarray(); $product_datas = $product_data_info['data']; if (empty($product_datas) && $page == 1) { //如果查询第一页为空,则直接返回 return true; } else if (empty($product_datas) && $page > 1) { //如果查询第二页为空,则直接返回,表示查询完毕,则处理数据清洗任务 $data_totle = count($product_datas); $index_number = 0; foreach ($product_datas as $key => $product_data) { $index_number = $key + 1; $message_data = ['company_id' => $company_id, 'page' => '1', 'limit' => '50', 'product_data' => $product_data, 'index_number' => $index_number, 'data_totle' => $data_totle]; LowPriceGoodsJobs::dispatch($message_data); // LowPriceGoodsJobs::dispatchSync($message_data); ViolationProductJobs::dispatch($message_data); // ViolationProductJobs::dispatchSync($message_data); ViolationCompanyJobs::dispatch($message_data); // ViolationCompanyJobs::dispatchSync($message_data); } } return true; } else { //合并数据 $product_datas = array_merge($product_datas, Cache::get($key_name)); Cache::put($key_name, $product_datas, 360); //缓存6分钟 //继续执行下一页 $message_data['page'] = $page + 1; $message_data['limit'] = $limit; ScrapeDataProductJobs::dispatch($message_data)->delay(now()->addSeconds(1)); } } catch (\Exception $e) { Log::info('job_error', '数据清洗-读取采集商品数据队列失败', ['data' => $this->message_data, 'error' => $e->getMessage()]); } } public function failed(\Throwable $exception) { Log::info('job_error', '数据清洗-读取采集商品数据队列完全失败', ['data' => $this->message_data, 'error' => $exception->getMessage()]); } }