Ai推荐文章领域队列
This commit is contained in:
124
application/api/job/RecommendArticleField.php
Normal file
124
application/api/job/RecommendArticleField.php
Normal file
@@ -0,0 +1,124 @@
|
||||
<?php
|
||||
namespace app\api\job;
|
||||
|
||||
use think\queue\Job;
|
||||
use app\common\QueueJob;
|
||||
use app\common\QueueRedis;
|
||||
use app\common\Article;
|
||||
class RecommendArticleField
|
||||
{
|
||||
private $logPath;
|
||||
private $oQueueJob;
|
||||
private $QueueRedis;
|
||||
private $maxRetries = 2;
|
||||
private $logBuffer = [];
|
||||
private $lastLogTime = 0;
|
||||
private $logMaxSize = 1048576; // 1MB (1*1024*1024) - 修正注释与值匹配
|
||||
private $lockExpire = 1800;
|
||||
private $completedExprie = 3600;
|
||||
const JSON_OPTIONS = JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR;
|
||||
|
||||
public function __construct()
|
||||
{
|
||||
$this->logPath = ROOT_PATH . 'public/queue_log/ArticleFiled_' . date('Ymd') . '.log';
|
||||
$this->oQueueJob = new QueueJob;
|
||||
$this->QueueRedis = QueueRedis::getInstance();
|
||||
$this->lastLogTime = time();
|
||||
// 确保日志目录存在
|
||||
$this->oQueueJob->ensureLogDirExists($this->logPath);
|
||||
}
|
||||
|
||||
public function fire(Job $job, $data)
|
||||
{
|
||||
$startTime = microtime(true);
|
||||
$this->oQueueJob->log("-----------队列任务开始-----------");
|
||||
|
||||
// 检查Redis连接状态
|
||||
if (!$this->QueueRedis->getConnectionStatus()) {
|
||||
$this->oQueueJob->log("Redis连接失败,10秒后重试");
|
||||
$job->release(10);
|
||||
$this->oQueueJob->flushLog();
|
||||
return;
|
||||
}
|
||||
|
||||
$iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];
|
||||
if (empty($iArticleId)) {
|
||||
$this->oQueueJob->log("无效的article_id,删除任务");
|
||||
$job->delete();
|
||||
$this->oQueueJob->flushLog();
|
||||
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);
|
||||
}
|
||||
}
|
||||
$this->oQueueJob->flushLog();
|
||||
return;
|
||||
}
|
||||
|
||||
$aParam = [
|
||||
'job_id' => $sRedisKey,
|
||||
'job_class' => $sClassName,
|
||||
'status' => 0,
|
||||
'create_time' => time(),
|
||||
'params' => json_encode($data, self::JSON_OPTIONS)
|
||||
];
|
||||
|
||||
$iLogId = $this->oQueueJob->addLog($aParam);
|
||||
if (!$iLogId) {
|
||||
$this->oQueueJob->log("日志创建失败,释放锁并删除任务:".json_encode($data, self::JSON_OPTIONS));
|
||||
$this->QueueRedis->releaseRedisLock($sRedisKey, $sRedisValue);
|
||||
$job->delete();
|
||||
$this->oQueueJob->flushLog();
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
$oArticle = new Article;
|
||||
$aResult = json_decode($oArticle->getAiField($data), true);
|
||||
$sMsg = empty($aResult['msg']) ? '内容生成成功' : $aResult['msg'];
|
||||
$this->oQueueJob->updateLog([
|
||||
'log_id' => $iLogId,
|
||||
'status' => 1,
|
||||
'update_time' => time(),
|
||||
'error' => $sMsg
|
||||
]);
|
||||
|
||||
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie);
|
||||
$job->delete();
|
||||
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId}");
|
||||
|
||||
} catch (\RuntimeException $e) {
|
||||
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
|
||||
} catch (\LogicException $e) {
|
||||
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
|
||||
} catch (\Exception $e) {
|
||||
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
|
||||
} finally {
|
||||
$executionTime = microtime(true) - $startTime;
|
||||
$this->oQueueJob->log("任务执行完成,耗时: " . number_format($executionTime, 4) . "秒");
|
||||
$this->oQueueJob->flushLog();
|
||||
gc_collect_cycles();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user