zm/tougao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

队列任务调整

Browse Source
This commit is contained in:
chengxl
2025-07-22 16:38:40 +08:00
parent 39c9bb78fa
commit 41063dfd06
12 changed files with 1133 additions and 944 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
application/api/job/createFieldForQueue.php
View File

@@ -1,162 +1,126 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\OpenAi;
use app\common\QueueJob;
use app\common\QueueRedis;
class createFieldForQueue
{
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;
private $logPath; // 日志路径
private $queueJob;
public function __construct()
{
$this->queueJob = new QueueJob;
// 初始化日志路径
$this->logPath = ROOT_PATH . 'public/queue_log/createFieldForQueue_' . date('Ymd') . '.log';
}
/**
* 安全写入日志（带文件锁）
*/
private function log($message)
{
$time = date('H:i:s');
$logMsg = "[$time] $message\n";
$fp = fopen($this->logPath, 'w');
if ($fp) {
flock($fp, LOCK_EX); // 排他锁防止并发写入冲突
fwrite($fp, $logMsg);
flock($fp, LOCK_UN);
fclose($fp);
}
}
// 任务日志添加
public function addLog($aParam = [])
{
//实例化
return $this->queueJob->addLog($aParam);
}
// 任务日志修改
public function updateLog($aParam = [])
{
return $this->queueJob->updateLog($aParam);
}
/**
* 根据错误信息获取重试延迟
*/
private function getRetryDelay($errorMsg)
{
$delayMap = [
'MySQL server has gone away' => 60,
'timeout' => 30,
'OpenAI' => 45,
'network' => 60
];
foreach ($delayMap as $keyword => $delay) {
if (strpos($errorMsg, $keyword) !== false) {
return $delay;
}
}
return 10; // 默认延迟
$this->oQueueJob = new QueueJob;
$this->QueueRedis = QueueRedis::getInstance();
$this->lastLogTime = time();
// 确保日志目录存在
$this->oQueueJob->ensureLogDirExists($this->logPath);
}
public function fire(Job $job, $data)
{
//日志
$this->log("-----------队列任务开始-----------");
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 获取文章ID
$iRedisId = empty($data['redis_id']) ? 0 : $data['redis_id'];
if (empty($iRedisId)) {
$this->log("无效的redis_id，删除任务");
$job->delete();
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
// 生成唯一任务标识
$sClassName = get_class($this);
$sRedisKey = "queue_job:{$sClassName}:{$iRedisId}";
$sRedisValue = uniqid() . '_' . getmypid(); // 增加进程ID确保唯一性
$iRedisId = empty($data['redis_id']) ? 0 : $data['redis_id'];
$sChunkIndex = empty($data['chunkIndex']) ? 0 : $data['chunkIndex'];
if (empty($iRedisId)) {
$this->oQueueJob->log("无效的redis_id，删除任务");
$job->delete();
$this->oQueueJob->flushLog();
return;
}
// 尝试获取Redis锁（原子操作）
$isLocked = $this->queueJob->setRedisLock($sRedisKey, $sRedisValue, 86400);
$sClassName = get_class($this);
$sRedisKey = "queue_job:{$sClassName}:{$iRedisId}:{$sChunkIndex}";
$sRedisValue = uniqid() . '_' . getmypid();
$lockExpire = $this->lockExpire;
$isLocked = $this->QueueRedis->startJob($sRedisKey, $sRedisValue, $lockExpire);
if (!$isLocked) {
$currentValue = $this->queueJob->getRedisValue($sRedisKey);
$this->log("任务已被锁定，避免重复执行 | 锁键: {$sRedisKey} | 锁值: {$currentValue}");
// 检查任务是否已超过最大重试次数
if ($job->attempts() >= 2) {
$this->log("任务超过最大重试次数，停止重试");
$jobStatus = $this->QueueRedis->getJobStatus($sRedisKey);
if (in_array($jobStatus, ['completed', 'failed'])) {
$this->oQueueJob->log("任务已完成或失败，删除任务 | 状态: {$jobStatus}");
$job->delete();
} else {
$delay = $this->getRetryDelay("任务已锁定");
$this->log("{$delay}秒后重试任务 | 重试次数: {$job->attempts()}");
$job->release($delay);
$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, // 0:处理中
'status' => 0,
'create_time' => time(),
'params' => json_encode($data, JSON_UNESCAPED_UNICODE)
'params' => json_encode($data, self::JSON_OPTIONS)
];
// 创建任务日志记录
$iLogId = $this->addLog($aParam);
if(!$iLogId) {
$this->log("日志创建失败，释放锁并删除任务:".json_encode($data));
$this->queueJob->releaseRedisLock($sRedisKey, $sRedisValue);
$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 {
// 执行核心任务
$oOpenAi = new OpenAi;
$aResult = json_decode($oOpenAi->createFieldForQueue($data), true);
$iStatus = empty($aResult['status']) ? 0 : $aResult['status'];
$sMsg = empty($aResult['msg']) ? '内容生成失败' : $aResult['msg'];
//更新任务状态
$aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg];
$this->updateLog($aParam);
//删除任务
$job->delete();
$this->log("任务执行成功，已删除任务 | 日志ID: {$iLogId}");
} catch (\Exception $e) {
//错误信息
$sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息
$sTrace = empty($e->getTraceAsString()) ? '' : $e->getTraceAsString();
$this->log("任务执行异常: {$sMsg} | 堆栈: {$sTrace}");
// 记录失败日志
$this->updateLog([
$sMsg = empty($aResult['msg']) ? '内容生成成功' : $aResult['msg'];
$this->oQueueJob->updateLog([
'log_id' => $iLogId,
'status' => 2,
'status' => 1,
'update_time' => time(),
'error' => $sMsg.':'.$sTrace,
'error' => $sMsg
]);
// 重试策略
$attempts = $job->attempts();
if ($attempts >= 2) {
$this->log("任务已重试{$attempts}次，停止重试");
$job->delete();
} else {
$delay = $this->getRetryDelay($sMsg);
$this->log("{$delay}秒后重试任务 | 重试次数: {$attempts}");
$job->release($delay);
}
$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 {
// 无论成功失败都释放锁（确保锁值匹配）
$releaseResult = $this->queueJob->releaseRedisLock($sRedisKey, $sRedisValue);
if (!$releaseResult) {
$this->log("释放锁失败 | 锁键: {$sRedisKey} | 锁值: {$sRedisValue}");
} else {
$this->log("成功释放锁 | 锁键: {$sRedisKey}");
}
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}