diff --git a/application/api/job/ArticleAiCreateContent.php b/application/api/job/ArticleAiCreateContent.php index 3e4704e..f774121 100644 --- a/application/api/job/ArticleAiCreateContent.php +++ b/application/api/job/ArticleAiCreateContent.php @@ -5,55 +5,93 @@ use app\api\controller\Aiarticle; use app\common\QueueJob; class ArticleAiCreateContent { + + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 文章AI内容生成 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/ArticleAiCreateContent_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; - try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '文章AI内容生成成功'; - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'ArticleAiCreateContent'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - //上传素材 - $oAiarticle = new Aiarticle; - $aResult = json_decode($oAiarticle->create($data),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '文章AI内容生成失败' : $aResult['msg']; - } + //获取文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + // 任务逻辑 + $sMsg = '文章AI内容生成成功'; + $iLogId = $this->addLog($aParam); + //生成内容 + $oAiarticle = new Aiarticle; + $aResult = json_decode($oAiarticle->create($data),true); + $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; + $sMsg = empty($aResult['msg']) ? '文章AI内容生成失败' : $aResult['msg']; //更新任务状态 $aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); - } catch (\Exception $e) { - - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); + $this->updateLog($aParam); + //删除任务 $job->delete(); + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + + } catch (\Exception $e) { + + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/RecommendReviewer.php b/application/api/job/RecommendReviewer.php index 16af1a6..c218479 100644 --- a/application/api/job/RecommendReviewer.php +++ b/application/api/job/RecommendReviewer.php @@ -5,92 +5,125 @@ use app\common\QueueJob; use app\common\Reviewer; class RecommendReviewer { + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 推荐审稿人任务 public function fire(Job $job, $data) - { + { - - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/RecommendReviewer_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; + //获取文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ + $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 try { + //添加任务日志 + $sMsg = '推荐审稿人任务处理成功'; + $iLogId = $this->addLog($aParam); - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '获取推荐审稿人信息成功'; - // 记录异常日志 - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'RecommendReviewer'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; + //获取推荐审稿人信息 + $aParam = ['article_id' => $iArticleId,'page' => 1,'size' => empty($data['size']) ? 5 : $data['size']]; + $oReviewer = new Reviewer; + $aResult = json_decode($oReviewer->recommend($aParam),true); + $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; + $sMsg = empty($aResult['msg']) ? '' : $aResult['msg']; - if (!empty($iArticleId)) { - - $aParam = ['article_id' => $iArticleId,'page' => 1,'size' => empty($data['size']) ? 5 : $data['size']]; - //获取推荐审稿人信息 - $oReviewer = new Reviewer; - $aResult = json_decode($oReviewer->recommend($aParam),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '' : $aResult['msg']; - //数量 - $iCount = empty($aResult['data']['total']) ? 0 : $aResult['data']['total']; - //推荐数量 - $iSize = empty($aResult['data']['size']) ? 0 : $aResult['data']['size']; - //判断是否给期刊管理者发邮件【数据库的审稿数量小于推荐数量】 - if($iCount < $iSize){ - $aSendEmailResult = json_decode($oReviewer->emailForEditor($aParam),true); + //处理数据 + $iCount = empty($aResult['data']['total']) ? 0 : $aResult['data']['total'];//数量 + $iSize = empty($aResult['data']['size']) ? 0 : $aResult['data']['size'];//推荐数量 + //判断是否给期刊管理者发邮件【数据库的审稿数量小于推荐数量】 + if($iCount < $iSize){ + $aSendEmailResult = json_decode($oReviewer->emailForEditor($aParam),true); + $sMsg .= ';'; + $sMsg .= empty($aSendEmailResult['msg']) ? '发送邮件入队成功' : $aSendEmailResult['msg']; + } + //推荐审稿人数据 + $aResult = empty($aResult['data']['lists']) ? [] : $aResult['data']['lists']; + if(empty($aResult)){ + $sMsg .= ';'; + $sMsg .= 'No qualified reviewers were selected'; + } + if(!empty($aResult)){ + $aParam = ['article_id' => $iArticleId,'reviewer_id' => array_column($aResult, 'reviewer_id')]; + $aResult = json_decode($oReviewer->add($aParam),true); + $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; + $sMsg .= ';'; + $sMsg .= empty($aResult['msg']) ? 'Reviewer data insertion failed' : $aResult['msg']; + //发送邀请审稿人审稿邮件 + if($iStatus == 1){ + $aResult = json_decode($oReviewer->email($aParam),true); + $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; $sMsg .= ';'; - $sMsg .= empty($aSendEmailResult['msg']) ? '发送邮件入队成功' : $aSendEmailResult['msg']; - } - //推荐审稿人数据 - $aResult = empty($aResult['data']['lists']) ? [] : $aResult['data']['lists']; - if(empty($aResult)){ - $sMsg .= ';'; - $sMsg .= 'No qualified reviewers were selected'; - } - if(!empty($aResult)){ - $aParam = ['article_id' => $iArticleId,'reviewer_id' => array_column($aResult, 'reviewer_id')]; - $aResult = json_decode($oReviewer->add($aParam),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg .= ';'; $sMsg .= empty($aResult['msg']) ? 'Reviewer data insertion failed' : $aResult['msg']; - //发送邀请审稿人审稿邮件 - if($iStatus == 1){ - $aResult = json_decode($oReviewer->email($aParam),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg .= ';'; - $sMsg .= empty($aResult['msg']) ? 'Reviewer data insertion failed' : $aResult['msg']; - } } } - $job->delete(); //更新任务状态 $aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg]; $oQueueJob->updateLog($aParam); - //记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); - } catch (\Exception $e) { - - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); + //删除任务 $job->delete(); + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { + + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/RelatedArticle.php b/application/api/job/RelatedArticle.php index cb2d641..122e7e0 100644 --- a/application/api/job/RelatedArticle.php +++ b/application/api/job/RelatedArticle.php @@ -5,57 +5,97 @@ use app\common\QueueJob; use app\common\JournalArticle; class RelatedArticle { + + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 相关文章发送邮件任务 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/RelatedArticle_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; - try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '获取相关文章信息成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'RelatedArticle'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - //上传素材 - $oJournalArticle = new JournalArticle; - $aResult = json_decode(JournalArticle::get($data),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '获取相关文章信息失败' : $aResult['msg']; - } - + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; + + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '关联文章任务处理成功'; + $iLogId = $this->addLog($aParam); + + //查询文章所关联的文章 + $oJournalArticle = new JournalArticle; + $aResult = json_decode(JournalArticle::get($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]; $oQueueJob->updateLog($aParam); - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); - + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/ReviewerScore.php b/application/api/job/ReviewerScore.php index ee1f275..045cb64 100644 --- a/application/api/job/ReviewerScore.php +++ b/application/api/job/ReviewerScore.php @@ -5,58 +5,96 @@ use app\common\QueueJob; use app\common\Reviewer; class ReviewerScore { + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 审稿人评分 public function fire(Job $job, $data) { - // 记录任务开始执行 + $sLogPath = ROOT_PATH.'public/queue_log/ReviewerScore_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); - - //获取任务ID - $iLogId = 0; - try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '给审稿人评分处理成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'ReviewerScore'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - - //审稿人评分 - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];//文章ID - $iReviewerId = empty($data['reviewer_id']) ? 0 : $data['reviewer_id'];//审稿人ID - $iArtRevId = empty($data['art_rev_id']) ? 0 : $data['art_rev_id'];//主键ID - if (!empty($iArticleId) && !empty($iArtRevId) && !empty($iArtRevId)) { - $aParam = ['article_id' => $iArticleId,'reviewer_id' => $iReviewerId,'art_rev_id' => $iArtRevId]; - $oReviewer = new Reviewer; - $aResult = json_decode($oReviewer->score($aParam),true); - $sMsg = empty($aResult['msg']) ? '给审稿人评分处理失败' : $aResult['msg']; - } + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); + //参数 + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];//文章ID + $iReviewerId = empty($data['reviewer_id']) ? 0 : $data['reviewer_id'];//审稿人ID + $iArtRevId = empty($data['art_rev_id']) ? 0 : $data['art_rev_id'];//主键ID + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId.'/'.$iReviewerId.'/'.$iArtRevId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '审稿人评分任务处理成功'; + $iLogId = $this->addLog($aParam); + + $aParam = ['article_id' => $iArticleId,'reviewer_id' => $iReviewerId,'art_rev_id' => $iArtRevId]; + $oReviewer = new Reviewer; + $aResult = json_decode($oReviewer->score($aParam),true); + $sMsg = empty($aResult['msg']) ? '给审稿人评分处理失败' : $aResult['msg']; //更新任务状态 $aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg]; $oQueueJob->updateLog($aParam); - // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); - + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); + }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/RevisionReviewer.php b/application/api/job/RevisionReviewer.php index 8005311..3faea07 100644 --- a/application/api/job/RevisionReviewer.php +++ b/application/api/job/RevisionReviewer.php @@ -5,60 +5,96 @@ use app\common\QueueJob; use app\common\Reviewer; class RevisionReviewer { + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 文章退修任务 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/RevisionReviewer_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; + //参数 + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];//文章ID + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ + $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '处理审稿人同意审稿但超时未审的数据成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'RevisionReviewer'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); + //添加任务日志 + $sMsg = '文章退修任务处理成功'; + $iLogId = $this->addLog($aParam); //获取符合条件的文章审稿人信息 - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - $aParam = ['article_id' => $iArticleId]; - $oReviewer = new Reviewer; - $aResult = json_decode($oReviewer->revisionForReviewer($aParam),true); - $sMsg = empty($aResult['msg']) ? '审稿人同意审稿但超时未审的数据失败' : $aResult['msg']; - } - - $job->delete(); - + $aParam = ['article_id' => $iArticleId]; + $oReviewer = new Reviewer; + $aResult = json_decode($oReviewer->revisionForReviewer($aParam),true); + $sMsg = empty($aResult['msg']) ? '审稿人同意审稿但超时未审的数据失败' : $aResult['msg']; + //更新任务状态 $aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg]; $oQueueJob->updateLog($aParam); - //记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); - } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); + //删除任务 $job->delete(); + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + + } catch (\Exception $e) { + + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } } - } \ No newline at end of file diff --git a/application/api/job/SendRelatedArticleEmail.php b/application/api/job/SendRelatedArticleEmail.php index 27c7643..15e068c 100644 --- a/application/api/job/SendRelatedArticleEmail.php +++ b/application/api/job/SendRelatedArticleEmail.php @@ -5,94 +5,131 @@ use app\common\QueueJob; use app\common\JournalArticle; class SendRelatedArticleEmail { - // 相关文章发送邮件任务 + + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + + // 关联文章任务 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/SendRelatedArticleEmail_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; - try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '发送邮件成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'SendRelatedArticleEmail'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; + //作者邮箱 + $email = empty($data['email']) ? '' : $data['email']; + //邮件主题 + $title = empty($data['title']) ? '' : $data['title']; + //发送来源 + $from_name = empty($data['from_name']) ? '' : $data['from_name']; + //邮件内容 + $content = empty($data['content']) ? '' : $data['content']; + //邮箱 + $memail = empty($data['memail']) ? '' : $data['memail']; + //密码 + $mpassword = empty($data['mpassword']) ? '' : $data['mpassword']; + //文章作者 + $article_author_id = empty($data['article_author_id']) ? 0 : $data['article_author_id']; + //关联文章ID + $related_article_id = empty($data['related_article_id']) ? 0 : $data['related_article_id']; + //期刊ID + $journal_id = empty($data['journal_id']) ? '' : $data['journal_id']; + //期刊issn + $journal_issn = empty($data['journal_issn']) ? '' : $data['journal_issn']; - //文章ID - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - //作者邮箱 - $email = empty($data['email']) ? '' : $data['email']; - //邮件主题 - $title = empty($data['title']) ? '' : $data['title']; - //发送来源 - $from_name = empty($data['from_name']) ? '' : $data['from_name']; - //邮件内容 - $content = empty($data['content']) ? '' : $data['content']; - //邮箱 - $memail = empty($data['memail']) ? '' : $data['memail']; - //密码 - $mpassword = empty($data['mpassword']) ? '' : $data['mpassword']; - //文章作者 - $article_author_id = empty($data['article_author_id']) ? 0 : $data['article_author_id']; - //关联文章ID - $related_article_id = empty($data['related_article_id']) ? 0 : $data['related_article_id']; - //期刊ID - $journal_id = empty($data['journal_id']) ? '' : $data['journal_id']; - //期刊issn - $journal_issn = empty($data['journal_issn']) ? '' : $data['journal_issn']; - //发送邮件 - if (!empty($iArticleId) && !empty($article_author_id) && !empty($related_article_id) && !empty($memail) && !empty($mpassword)) { - - //查询是否发送过邮件 - $oJournalArticle = new JournalArticle; - $aLog = json_decode($oJournalArticle::getLog(['article_id' => $iArticleId,'article_author_id' => $article_author_id,'related_article_id' => $related_article_id,'is_success' => 1]),true); - $sMsg = '邮件已发送:'.json_encode($aLog['data']); - if(empty($aLog['data'])){ - $aResult = sendEmail($email,$title,$from_name,$content,$memail,$mpassword); - $iStatus = empty($aResult['status']) ? 1 : $aResult['status']; - $iIsSuccess = 2; - $sMsg = empty($aResult['data']) ? '失败' : $aResult['data']; - if($iStatus == 1){ - $iIsSuccess = 1; - $sMsg = '成功'; - } - - //记录邮件发送日志 - $aEmailLog = ['article_id' => $iArticleId,'article_author_id' => $article_author_id,'related_article_id' => $related_article_id,'email' => $email,'content' => $content,'create_time' => time(),'is_success' => $iIsSuccess,'journal_id' => $journal_id,'journal_issn' => $journal_issn,'msg' => $sMsg]; - //添加邮件发送日志 - $iId = JournalArticle::addLog($aEmailLog); - } - } - + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId.'/'.$related_article_id.'/'.$article_author_id.'/'.$email; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '关联文章任务处理成功'; + $iLogId = $this->addLog($aParam); + + //查询是否发送过邮件 + $oJournalArticle = new JournalArticle; + $aLog = json_decode($oJournalArticle::getLog(['article_id' => $iArticleId,'article_author_id' => $article_author_id,'related_article_id' => $related_article_id,'is_success' => 1]),true); + $sMsg = '邮件已发送'; + if(empty($aLog['data'])){ + $aResult = sendEmail($email,$title,$from_name,$content,$memail,$mpassword); + $iStatus = empty($aResult['status']) ? 1 : $aResult['status']; + $iIsSuccess = 2; + $sMsg = empty($aResult['data']) ? '失败' : $aResult['data']; + if($iStatus == 1){ + $iIsSuccess = 1; + $sMsg = '成功'; + } + + //记录邮件发送日志 + $aEmailLog = ['article_id' => $iArticleId,'article_author_id' => $article_author_id,'related_article_id' => $related_article_id,'email' => $email,'content' => $content,'create_time' => time(),'is_success' => $iIsSuccess,'journal_id' => $journal_id,'journal_issn' => $journal_issn,'msg' => $sMsg]; + //添加邮件发送日志 + $iId = JournalArticle::addLog($aEmailLog); + } + //更新任务状态 $aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg]; $oQueueJob->updateLog($aParam); - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); - + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/SendReviewEmail.php b/application/api/job/SendReviewEmail.php index cfc04dd..880a629 100644 --- a/application/api/job/SendReviewEmail.php +++ b/application/api/job/SendReviewEmail.php @@ -5,92 +5,130 @@ use app\common\QueueJob; use app\common\Reviewer; class SendReviewEmail { + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 发送审稿邀请邮件任务 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/SendReviewEmail_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------'."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; + //作者邮箱 + $email = empty($data['email']) ? '' : $data['email']; + //邮件主题 + $title = empty($data['title']) ? '' : $data['title']; + //发送来源 + $from_name = empty($data['from_name']) ? '' : $data['from_name']; + //邮件内容 + $content = empty($data['content']) ? '' : $data['content']; + //邮箱 + $memail = empty($data['memail']) ? '' : $data['memail']; + //密码 + $mpassword = empty($data['mpassword']) ? '' : $data['mpassword']; + //审稿记录表主键ID + $art_rev_id = empty($data['art_rev_id']) ? 0 : $data['art_rev_id']; + //审稿人ID + $reviewer_id = empty($data['reviewer_id']) ? 0 : $data['reviewer_id']; + //邮件类型 + $type = empty($data['type']) ? 1 : $data['type']; + + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId.'/'.$art_rev_id.'/'.$reviewer_id.'/'.$email; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ + $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '发送邮件成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'SendReviewEmail'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - //文章ID - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - //作者邮箱 - $email = empty($data['email']) ? '' : $data['email']; - //邮件主题 - $title = empty($data['title']) ? '' : $data['title']; - //发送来源 - $from_name = empty($data['from_name']) ? '' : $data['from_name']; - //邮件内容 - $content = empty($data['content']) ? '' : $data['content']; - //邮箱 - $memail = empty($data['memail']) ? '' : $data['memail']; - //密码 - $mpassword = empty($data['mpassword']) ? '' : $data['mpassword']; - //审稿记录表主键ID - $art_rev_id = empty($data['art_rev_id']) ? 0 : $data['art_rev_id']; - //审稿人ID - $reviewer_id = empty($data['reviewer_id']) ? 0 : $data['reviewer_id']; - //邮件类型 - $type = empty($data['type']) ? 1 : $data['type']; + //添加任务日志 + $sMsg = '发送审稿邀请邮件任务处理成功'; + $iLogId = $this->addLog($aParam); //发送邮件 - if (!empty($email) && !empty($memail) && !empty($mpassword)) { - - //查询是否发送过邮件 - $oReviewer = new Reviewer; - if($type != 3){ - $aLog = json_decode($oReviewer->getLog(['article_id' => $iArticleId,'art_rev_id' => $art_rev_id,'reviewer_id' => $reviewer_id,'is_success' => 1,'type' => $type]),true); - $aLog = empty($aLog['data']) ? [] : $aLog['data']; - $sMsg = '邮件已发送:'.json_encode($aLog); - } - if(empty($aLog)){ - $aResult = sendEmail($email,$title,$from_name,$content,$memail,$mpassword); - $iStatus = empty($aResult['status']) ? 1 : $aResult['status']; - $iIsSuccess = 2; - $sMsg = empty($aResult['data']) ? '失败' : $aResult['data']; - if($iStatus == 1){ - $iIsSuccess = 1; - $sMsg = '成功'; - } - //记录邮件发送日志 - $aEmailLog = ['article_id' => $iArticleId,'art_rev_id' => $art_rev_id,'reviewer_id' => $reviewer_id,'type' => $type,'email' => $email,'content' => $content,'create_time' => time(),'is_success' => $iIsSuccess,'msg' => $sMsg]; - //添加邮件发送日志 - $iId = $oReviewer->addLog($aEmailLog); + //查询是否发送过邮件 + $oReviewer = new Reviewer; + if($type != 3){ + $aLog = json_decode($oReviewer->getLog(['article_id' => $iArticleId,'art_rev_id' => $art_rev_id,'reviewer_id' => $reviewer_id,'is_success' => 1,'type' => $type]),true); + $aLog = empty($aLog['data']) ? [] : $aLog['data']; + $sMsg = '邮件已发送:'.json_encode($aLog); + } + if(empty($aLog)){ + $aResult = sendEmail($email,$title,$from_name,$content,$memail,$mpassword); + $iStatus = empty($aResult['status']) ? 1 : $aResult['status']; + $iIsSuccess = 2; + $sMsg = empty($aResult['data']) ? '失败' : $aResult['data']; + if($iStatus == 1){ + $iIsSuccess = 1; + $sMsg = '成功'; } + //记录邮件发送日志 + $aEmailLog = ['article_id' => $iArticleId,'art_rev_id' => $art_rev_id,'reviewer_id' => $reviewer_id,'type' => $type,'email' => $email,'content' => $content,'create_time' => time(),'is_success' => $iIsSuccess,'msg' => $sMsg]; + //添加邮件发送日志 + $iId = $oReviewer->addLog($aEmailLog); } - $job->delete(); - //更新任务状态 $aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg]; $oQueueJob->updateLog($aParam); - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/WechatDraft.php b/application/api/job/WechatDraft.php index b69b78f..1b62a89 100644 --- a/application/api/job/WechatDraft.php +++ b/application/api/job/WechatDraft.php @@ -5,57 +5,95 @@ use app\api\controller\Aiarticle; use app\common\QueueJob; class WechatDraft { + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 上传文章到草稿箱任务入口 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/WechatDraft_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); + + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - //获取任务ID - $iLogId = 0; - try { - - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '上传草稿箱成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'WechatDraft'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - //上传素材 - $oAiarticle = new Aiarticle; - $aResult = json_decode($oAiarticle->syncWechat($data),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '上传草稿箱失败' : $aResult['msg']; - } - + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '上传草稿箱处理成功'; + $iLogId = $this->addLog($aParam); + + //上传草稿箱 + $oAiarticle = new Aiarticle; + $aResult = json_decode($oAiarticle->syncWechat($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]; $oQueueJob->updateLog($aParam); - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end---------'."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/WechatDraftPublish.php b/application/api/job/WechatDraftPublish.php index cdf2c3e..a05509d 100644 --- a/application/api/job/WechatDraftPublish.php +++ b/application/api/job/WechatDraftPublish.php @@ -5,57 +5,97 @@ use app\api\controller\Aiarticle; use app\common\QueueJob; class WechatDraftPublish { + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 发布草稿箱任务入口 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/WechatDraftPublish_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); - //获取任务ID - $iLogId = 0; - try { + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '草稿箱发布成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'WechatDraftPublish'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - //上传素材 - $oAiarticle = new Aiarticle; - $aResult = json_decode($oAiarticle->publishDraft($data),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '草稿箱发布失败' : $aResult['msg']; - } - + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '草稿箱发布任务处理成功'; + $iLogId = $this->addLog($aParam); + + + //发布草稿箱 + $oAiarticle = new Aiarticle; + $aResult = json_decode($oAiarticle->publishDraft($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]; $oQueueJob->updateLog($aParam); - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end-----------' . json_encode($data)."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/WechatMaterial.php b/application/api/job/WechatMaterial.php index c7a0c28..93e9a99 100644 --- a/application/api/job/WechatMaterial.php +++ b/application/api/job/WechatMaterial.php @@ -5,59 +5,96 @@ use app\api\controller\Aiarticle; use app\common\QueueJob; class WechatMaterial { + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 上传素材任务入口 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/WechatMaterial_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); + + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - //获取任务ID - $iLogId = 0; - try { - - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '上传素材成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'WechatMaterial'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - //上传素材 - $oAiarticle = new Aiarticle; - $aResult = json_decode($oAiarticle->uploadMaterial($data),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '上传素材失败' : $aResult['msg']; - } - + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '上传素材任务处理成功'; + $iLogId = $this->addLog($aParam); + + //上传素材 + $oAiarticle = new Aiarticle; + $aResult = json_decode($oAiarticle->uploadMaterial($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]; $oQueueJob->updateLog($aParam); - - // // 记录日志 - file_put_contents($sLogPath,'-----------Queue job end-----------' . json_encode($data)."\n\n\n",FILE_APPEND); - } catch (\Exception $e) { - - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); + + //删除任务 $job->delete(); + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + + } catch (\Exception $e) { + + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 } diff --git a/application/api/job/WechatQueryStatus.php b/application/api/job/WechatQueryStatus.php index 4cb442b..dfa7442 100644 --- a/application/api/job/WechatQueryStatus.php +++ b/application/api/job/WechatQueryStatus.php @@ -5,56 +5,96 @@ use app\api\controller\Aiarticle; use app\common\QueueJob; class WechatQueryStatus { + + // 最多重试1次 + public $tries = 1; + + // 任务日志添加 + public function addLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + $iLogId = $oQueueJob->addLog($aParam); + return $iLogId; + } + + // 任务日志修改 + public function updateLog($aParam = []) + { + //实例化 + $oQueueJob = new QueueJob; + return $oQueueJob->updateLog($aParam); + } + // 草稿箱文章发布状态任务入口 public function fire(Job $job, $data) { - // 记录任务开始执行 $sLogPath = ROOT_PATH.'public/queue_log/WechatQueryStatus_'.date('Ymd').'.log'; - file_put_contents($sLogPath,'-----------Queue job started-----------' . json_encode($data)."\n",FILE_APPEND); + $sTime = date('H:i:s'); + file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------'); + + //文章ID + $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - //获取任务ID - $iLogId = 0; - try { - //实例化 - $oQueueJob = new QueueJob; - $sMsg = '查询草稿箱文章发布状态成功'; - - $aJob = empty($job->getRawBody()) ? [] : json_decode($job->getRawBody(), true); - $aParam = [ - 'job_id' => empty($aJob['id']) ? 'WechatQueryStatus'.rand(100, 999) : $aJob['id'], - 'job_class' => get_class($this), - 'status' => 0, - 'create_time' => time(), - 'params' => json_encode($data, JSON_UNESCAPED_UNICODE) - ]; - $iLogId = $oQueueJob->addLog($aParam); - // 步骤1:上传素材(图片) - $iArticleId = empty($data['article_id']) ? 0 : $data['article_id']; - if (!empty($iArticleId)) { - //上传素材 - $oAiarticle = new Aiarticle; - $aResult = json_decode($oAiarticle->queryStatus($data),true); - $iStatus = empty($aResult['status']) ? 0 : $aResult['status']; - $sMsg = empty($aResult['msg']) ? '上传草稿箱失败' : $aResult['msg']; - } - + //获取方法名 + $sClassName = get_class($this); + // 检查任务是否已处理(基于业务唯一标识) + $sRedisKey = $sClassName.'/'.$iArticleId; + $sRedisKey = md5($sRedisKey); + //判断Redis是否存在 + $oQueueJob = new QueueJob; + $result = $oQueueJob->setRedisLabel(['redis_key' => $sRedisKey]); + if(empty($result)){ $job->delete(); + file_put_contents($sLogPath,'-----------Queue job already:'.$sTime.'-----------'); + return; + } + + //任务数组 + $aParam = [ + 'job_id' => $sRedisKey, + 'job_class' => $sClassName, + 'status' => 0, + 'create_time' => time(), + 'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE) + ]; + //执行任务 + try { + //添加任务日志 + $sMsg = '查询草稿箱文章发布处理成功'; + $iLogId = $this->addLog($aParam); + + // 查询状态 + $oAiarticle = new Aiarticle; + $aResult = json_decode($oAiarticle->queryStatus($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]; $oQueueJob->updateLog($aParam); - // // 记录错误日志 - file_put_contents($sLogPath,'-----------Queue job end-----------' . json_encode($data)."\n\n\n",FILE_APPEND); + + //删除任务 + $job->delete(); + + file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------'); + } catch (\Exception $e) { - //实例化 - $oQueueJob = new QueueJob; - //更新任务状态 - $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); - $aParam = ['log_id' => $iLogId,'status' => 2,'update_time' => time(),'error' => $sMsg]; - $oQueueJob->updateLog($aParam); - $job->delete(); + // 2. 记录失败日志 + $aParam['status'] = 2; // 标记状态为"失败" + $sMsg = empty($e->getMessage()) ? '任务出错' : $e->getMessage(); // 错误信息 + $aParam['error'] = $sMsg; + $this->addLog($aParam); // 调用日志记录方法 + if ($job->attempts() > $this->tries) { + //如果任务尝试次数超过最大重试次数 + $job->delete(); // 删除任务,不再重试 + } else { + // 3. 如果尝试次数未超过最大重试次数,释放任务回队列 + $job->release(30); // 30秒后重新尝试执行任务 + } + file_put_contents($sLogPath,'-----------Queue job error:'.$sMsg.'-----------'.$sTime); }finally { gc_collect_cycles(); // 强制垃圾回收 }