oQueueJob = new QueueJob; $this->QueueRedis = QueueRedis::getInstance(); } public function fire(Job $job, $data) { $startTime = microtime(true); $this->oQueueJob->log("-----------队列任务开始-----------"); // 检查数据库连接 if (!$this->oQueueJob->checkDbConnection(true)) { $this->oQueueJob->log("数据库连接失败,无法执行任务,10秒后重试"); $job->release(10); return; } // 检查Redis连接状态 if (!$this->QueueRedis->getConnectionStatus()) { $this->oQueueJob->log("Redis连接失败,10秒后重试"); $job->release(10); return; } // 获取文章ID $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; if (empty($iArticleId)) { $this->oQueueJob->log("无效的article_id,删除任务"); $job->delete(); return; } // 生成唯一任务标识 $sClassName = get_class($this); $sRedisKey = "queue_job:{$sClassName}:{$iArticleId}"; $sRedisValue = uniqid() . '_' . getmypid(); $lockExpire = $this->lockExpire; $isLocked = $this->QueueRedis->startJob($sRedisKey, $sRedisValue, $lockExpire); if (!$isLocked) { $jobStatus = $this->QueueRedis->getJobStatus($sRedisKey); if (in_array($jobStatus, ['completed', 'failed'])) { $this->oQueueJob->log("任务已完成或失败,删除任务 | 状态: {$jobStatus}"); $job->delete(); } else { $attempts = $job->attempts(); if ($attempts >= $this->maxRetries) { $this->oQueueJob->log("超过最大重试次数,停止重试"); $job->delete(); } else { $lockTtl = $this->QueueRedis->getLockTtl($sRedisKey); $delay = $lockTtl > 0 ? $lockTtl + 5 : 30; $this->oQueueJob->log("锁竞争,{$delay}秒后重试({$attempts}/{$this->maxRetries})"); $job->release($delay); } } return; } try { // 执行核心任务前再次检查连接 $result = $this->oQueueJob->checkDbConnection(); if (!$result) { throw new \RuntimeException("数据库连接异常,无法执行核心任务"); } //获取符合条件的文章审稿人信息 $aParam = ['article_id' => $iArticleId]; $oReviewer = new Reviewer; $aResult = json_decode($oReviewer->revisionForReviewer($aParam),true); $sMsg = empty($aResult['msg']) ? '审稿人同意审稿但超时未审的数据失败' : $aResult['msg']; //更新日志 $this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue); $job->delete(); $this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}"); } catch (\RuntimeException $e) { $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job); } catch (\LogicException $e) { $this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue,$job); } catch (\Exception $e) { $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job); } finally { $executionTime = microtime(true) - $startTime; $this->oQueueJob->log("任务执行完成,耗时: " . number_format($executionTime, 4) . "秒"); gc_collect_cycles(); } } }