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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
wangjinlei
2025-07-25 17:28:13 +08:00
parent ca9d2a7412 3bace093ff
commit 0506fa03f3
21 changed files with 2595 additions and 1762 deletions
Download Patch File Download Diff File Expand all files Collapse all files

337
application/api/controller/Aiarticle.php
View File

@@ -18,9 +18,6 @@ class Aiarticle extends Base
{
protected $aLogo = ['media_id' => 'Cn8zlXvVB5DwjcA9h40z9fprHDoc3Jqv97SwrInpmyYiilkeRdKvpD63cWqTYHfz','url' => 'http://mmbiz.qpic.cn/mmbiz_jpg/QHFVW13lONaQJxK9QbHU9CtrvTS2ModZnUyeAvuVN67t8XP85DxVJwDJf2YxCTalrsr17jS080xM6xQv5yGiaEQ/0?wx_fmt=jpeg'];//默认头像
//数据表必填字段[ai_article]
protected $aAiFileds = ['article_id','title_english','title_chinese','journal_issn','covered','digest','research_result','content','highlights','discussion','prospect','research_background','discussion_results','research_method','overview','summary'];
//期刊接口地址
protected $sJournalUrl = 'http://journalapi.tmrjournals.com/public/index.php';//'http://zmzm.journal.dev.com'; //
//期刊官网
@@ -53,59 +50,10 @@ class Aiarticle extends Base
}
/**
* @title 请求OPENAI
* @title 生成公微
* @param aSearch array 模版内容
* @param iTemplateId 模版ID
*/
private function _createContentForOpenAI($aSearch = [],$sArticleType = 'default',$isTranslate = 2){
if(empty($aSearch) || empty($sArticleType)){
return json_encode(array('status' => 2,'msg' => 'Please select a template or enter the content you want to consult'));
}
//组织参数
if(in_array($sArticleType, ['Mini Review','Review'])){
$sArticleType = 'Review';
}else{
$sArticleType = 'default';
}
if($isTranslate == 1){
$sArticleType = 1;
}
//获取问答内容
$oOpenAi = new OpenAi;
if($sArticleType == 'default'){
$aMessage = $oOpenAi->buildDefaultPrompt($aSearch);
if(empty($aMessage)){
return json_encode(['status' => 5,'msg' => 'AI Q&A content not obtained']);
}
$aParam = ['messages' => $aMessage,'model' => empty($aParam['api_model']) ? 'gpt-4.1' : $aParam['api_model'],'redis_id' => empty($aSearch['article_id']) ? 0 : $aSearch['article_id']];
$aResult = json_decode($oOpenAi->createWechatContent($aParam),true);
}
if($sArticleType == 'Review'){
$aMessage = $oOpenAi->buildReviewPrompt($aSearch);
if(empty($aMessage)){
return json_encode(['status' => 5,'msg' => 'AI Q&A content not obtained']);
}
$aParam = ['messages' => $aMessage,'model' => empty($aParam['api_model']) ? 'gpt-4.1' : $aParam['api_model'],'redis_id' => empty($aSearch['article_id']) ? 0 : $aSearch['article_id']];
$aResult = json_decode($oOpenAi->createWechatContent($aParam),true);
}
if($sArticleType == 1){
$aMessage = $oOpenAi->buildTranslatePrompt($aSearch);
if(empty($aMessage)){
return json_encode(['status' => 5,'msg' => 'AI Q&A content not obtained']);
}
$aParam = ['messages' => $aMessage,'model' => empty($aParam['api_model']) ? 'gpt-4.1' : $aParam['api_model']];
$aResult = json_decode($oOpenAi->curlOpenAI($aParam),true);
}
//请求OPENAI接口
$iStatus = empty($aResult['status']) ? 0 : $aResult['status'];
if($iStatus != 1){
return json_encode($aResult);
}
return json_encode(array('status' => 1,'msg' => 'Content is being generated, please wait'));
}
public function create($aParam = []){
//获取参数
@@ -157,8 +105,7 @@ class Aiarticle extends Base
return json_encode(['status' => 5,'msg' => 'The data has been generated, please proceed with the next steps']);
}
//请求OPENAI
$aResult = json_decode($this->createForOpenAi($aArticleContent,$aAiArticle),true);
return json_encode($aResult);
return $this->createForOpenAi($aArticleContent,$aAiArticle);
}
@@ -198,12 +145,17 @@ class Aiarticle extends Base
$aSearch['{#title_chinese#}'] = $this->basic_html_filter($aArticle['title']);
$aSearch['{#journal_content#}'] = $this->basic_html_filter($aJournal['journal_content'] ?? '');
$aSearch['{#journal_content#}'] = trim(trim($aSearch['{#journal_content#}'],'.'),'。');
//写入Redis 标识已经开始请求OPENAI处理
$sArticleType = empty($aArticle['type']) ? 'default' : $aArticle['type'];
$aSearch['article_id'] = $iArticleId;
$aResult = $this->_createContentForOpenAI($aSearch,$sArticleType);
return $aResult;
$aSearch['prompt_article_type'] = empty($aArticle['type']) ? 'default' : $aArticle['type'];
//获取问答内容
$oOpenAi = new OpenAi;
$aMessage = $oOpenAi->buildWechatPrompt($aSearch);
if(empty($aMessage)){
return json_encode(['status' => 5,'msg' => 'AI Q&A content not obtained']);
}
//请求OPENAI接口
$aParam = ['messages' => $aMessage,'model' => empty($aParam['api_model']) ? 'gpt-4.1' : $aParam['api_model'],'redis_id' => empty($aSearch['article_id']) ? 0 : $aSearch['article_id']];
return $oOpenAi->createWechatContent($aParam);
}
/**
@@ -277,7 +229,7 @@ class Aiarticle extends Base
//是否查询作者 1是2否
$iSelectAuthor = empty($aParam['is_select_author']) ? 2 : $aParam['is_select_author'];
//查询AI生成的文章内容
$aWhere = ['is_delete' => 2];
$aWhere = ['is_delete' => 2,'is_generate' => 1];
if(!empty($iArticleId)){
$aWhere['article_id'] = $iArticleId;
}
@@ -390,118 +342,74 @@ class Aiarticle extends Base
//获取参数
$aParam = empty($aParam) ? $this->request->post() : $aParam;
//主键ID
$iAiArticleId = empty($aParam['ai_article_id']) ? 0 : $aParam['ai_article_id'];
//文章ID
$iArticleId = empty($aParam['article_id']) ? 0 : $aParam['article_id'];
//查询内容是否存在
$aWhere = ['is_delete' => 2];
if(empty($iArticleId) && empty($iArticleId)){
return json_encode(['status' => 2,'msg' => 'Please select the article to be modified']);
//更新AI文章内容
$oArticle = new \app\common\Article;
$aResult = json_decode($oArticle->updateAiArticle($aParam),true);
$iStatus = empty($aResult['status']) ? 0 : $aResult['status'];
if($iStatus != 1){
return json_encode($aResult);
}
if(!empty($iArticleId)){
$aWhere['article_id'] = $iArticleId;
}
if(!empty($iAiArticleId)){
$aWhere['ai_article_id'] = $iAiArticleId;
}
$aAiArticle = Db::name('ai_article')->where($aWhere)->find();
if(empty($aAiArticle)){
return json_encode(['status' => 3,'msg' => 'he article content of WeChat official account has not been generated']);
}
//必填参数验证
$aFields = $this->aAiFileds;
$sFiled = '';
$aUpdateParam = [];
foreach($aFields as $val){
if(!isset($aParam[$val])){
continue;
}
if(is_array($aParam[$val])){
$aParam[$val] = implode(";",$aParam[$val]);
}
$aUpdateParam[$val] = empty($aParam[$val]) ? '' : addslashes(strip_tags($aParam[$val]));
}
if(empty($aUpdateParam)){
return json_encode(['status' => 1,'msg' => 'No data currently being processed']);
}
Db::startTrans();
//执行入库
$aUpdateParam['update_time'] = time();
$aUpdateParam['is_generate'] = 1;
$result = Db::name('ai_article')->where('ai_article_id',$iAiArticleId)->limit(1)->update($aUpdateParam);
if($result === false){
$aResult = json_encode(['status' => 3,'msg' => 'UPDATEING AI article failed']);
}
$aResult['data'] = $aUpdateParam;
//更新作者信息
$aAuthorList = empty($aParam['author_list']) ? []: $aParam['author_list'];
$aAuthorList = is_array( $aAuthorList) ? $aAuthorList: json_decode($aAuthorList,true);
if(!empty($aAuthorList)){
$aAuthorList = array_column($aAuthorList, null,'email');
//根据邮箱查询作者信息
$aEmail = array_keys($aAuthorList);
if(empty($aAuthorList)){
return json_encode($aResult);
}
$aUserInfo = $aUserReviewer = [];
if(!empty($aEmail)){
$aUserParam = ['email' => ['in',$aEmail]];
$aUserInfo = Db::name('user')->field('email,user_id')->where($aUserParam)->select();
if(!empty($aUserInfo)){
$aUserId = array_column($aUserInfo, 'user_id');
//查询用户附表
$aWhere = ['reviewer_id' => ['in',$aUserId]];
$aUserReviewer = Db::name('user_reviewer_info')->where($aWhere)->column('reviewer_id');
//根据邮箱查询作者信息
$aAuthorList = array_column($aAuthorList, null,'email');
$aEmail = array_keys($aAuthorList);
if(empty($aEmail)){
return json_encode($aResult);
}
$aUserParam = ['email' => ['in',$aEmail]];
$aUserInfo = Db::name('user')->field('email,user_id')->where($aUserParam)->select();
if(empty($aUserInfo)){
return json_encode($aResult);
}
//查询用户附表
$aUserId = array_column($aUserInfo, 'user_id');
$aWhere = ['reviewer_id' => ['in',$aUserId]];
$aUserReviewer = Db::name('user_reviewer_info')->where($aWhere)->column('reviewer_id');
//更新用户信息
Db::startTrans();
foreach ($aUserInfo as $key => $value) {
$aUser = empty($aAuthorList[$value['email']]) ? [] : $aAuthorList[$value['email']];
if(empty($aUser)){
continue;
}
//更新作者名字
if(isset($aUser['author_name'])){
$aUpdate = ['localname' => $aUser['author_name']];
Db::name('user')->where('user_id',$value['user_id'])->update($aUpdate);
}
//更新作者简介和单位
$aUpdateReviewer = [];
if(isset($aUser['technical'])){
$aUpdateReviewer['technical'] = $aUser['technical'];
}
if(isset($aUser['introduction'])){
$aUpdateReviewer['introduction'] = $aUser['introduction'];
}
if(isset($aUser['company'])){
$aUpdateReviewer['company'] = $aUser['company'];
}
if(isset($aUpdateReviewer)){
if(in_array($value['user_id'], $aUserReviewer)){//更新
Db::name('user_reviewer_info')->where('reviewer_id',$value['user_id'])->limit(1)->update($aUpdateReviewer);
}
if(!in_array($value['user_id'], $aUserReviewer)){//插入
$aUpdateReviewer['reviewer_id'] = $value['user_id'];
Db::name('user_reviewer_info')->insert($aUpdateReviewer);
}
}
//更新用户信息
if(!empty($aUserInfo)){
foreach ($aUserInfo as $key => $value) {
$aUser = empty($aAuthorList[$value['email']]) ? [] : $aAuthorList[$value['email']];
if(empty($aUser)){
continue;
}
//更新作者名字
if(isset($aUser['author_name'])){
$aUpdate = ['localname' => $aUser['author_name']];
Db::name('user')->where('user_id',$value['user_id'])->update($aUpdate);
}
//更新作者简介和单位
$aUpdateReviewer = [];
if(isset($aUser['technical'])){
$aUpdateReviewer['technical'] = $aUser['technical'];
}
if(isset($aUser['introduction'])){
$aUpdateReviewer['introduction'] = $aUser['introduction'];
}
if(isset($aUser['company'])){
$aUpdateReviewer['company'] = $aUser['company'];
}
if(isset($aUpdateReviewer)){
if(in_array($value['user_id'], $aUserReviewer)){//更新
Db::name('user_reviewer_info')->where('reviewer_id',$value['user_id'])->limit(1)->update($aUpdateReviewer);
}
if(!in_array($value['user_id'], $aUserReviewer)){//插入
$aUpdateReviewer['reviewer_id'] = $value['user_id'];
Db::name('user_reviewer_info')->insert($aUpdateReviewer);
}
}
}
}
}
Db::commit();
$aResult['data'] = $aUpdateParam;
$aResult['data'] = $aParam;
return json_encode($aResult);
}
@@ -695,11 +603,11 @@ class Aiarticle extends Base
//处理通讯作者信息数据
$aAuthorInfo = json_decode($this->dealTemplateAuthor($aAuthor,$sTemplatePath),true);
$aAuthorInfo = empty($aAuthorInfo['data']) ? [] : $aAuthorInfo['data'];
$aSearch['{###author_summary###}'] = empty($aAuthorInfo['author_info']) ? '' : $aAuthorInfo['author_info'];
$aSearch['{###author_string###}'] = empty($aAuthorInfo['author_string']) ? '' : '，通讯作者为'.$aAuthorInfo['author_string'];
//处理往期推荐数据
$aPreviousRecommend = json_decode($this->dealTemplatePreviousRecommend($aArticle,$sTemplatePath),true);
$aSearch['{###previous_recommend_summary###}'] = empty($aPreviousRecommend['data']) ? '' : $aPreviousRecommend['data'];
@@ -741,12 +649,6 @@ class Aiarticle extends Base
}
$aSearch['{###jabbr###}'] = trim(trim($aSearch['{###jabbr###}'],'.'),'。');
// $aSearch['{###free_year###}'] = date('Y');
// $aSearch['{###covered###}'] = '内容涵盖'.str_replace('研究涵盖', '', $aSearch['{###covered###}']);
if(!empty($aSearch['{###covered###}']) && strpos($aSearch['{###covered###}'], '涵盖') === false){
$aSearch['{###covered###}'] = '内容涵盖'.str_replace('研究涵盖', '', $aSearch['{###covered###}']);
}
//期刊封面
$aSearch['{###journal_icon_info###}'] = $aSearch['{###journal_icon_details###}'] = '';
if(!empty($sJournalIcon)){
@@ -761,7 +663,8 @@ class Aiarticle extends Base
$aSearch['{###journal_icon_details###}'] = str_replace('{###journal_icon###}', $aAiArticle['journal_icon']??'', $sTemplateJournalInfo);
}
}
$aSearch['{###covered###}'] = empty($aSearch['{###covered###}']) ? '' : '内容涵盖'.$aSearch['{###covered###}'];
$sTemplate = str_replace(array_keys($aSearch), array_values($aSearch), $sTemplate);
// $this->writeFile($sTemplate,ROOT_PATH.'public/template_info.html');
return json_encode(['status' => 1, 'msg' => '模版生成成功','data' => ['template' => $sTemplate]]);
@@ -788,58 +691,70 @@ class Aiarticle extends Base
}
$sAuthorTemplate = file_get_contents($sAuthorTemplatePathAuthor);
//处理模版变量
$aAuthorSerachInfo = [];
$sAuthorInfo = '';
//默认头像
$aLogo = $this->aLogo;
//请求AI翻译
$sContent = '';
$aSearch['{#content#}'] = json_encode($aAuthor);
$aResult = json_decode($this->_createContentForOpenAI($aSearch,1,1),true);
$aContent = empty($aResult['data']) ? [] : array_column($aResult['data'], null,'email');
//数据处理
$aTranslate = $aAuthorSerachInfo = $aTranslateAuthorList = [];
//通讯作者信息
$sAuthorInfo = '';
foreach ($aAuthor as $key => $value) {
//请求AI翻译
$aInfo = empty($aContent[$value['email']]) ? [] : $aContent[$value['email']];
//替换模版
$aAuthorSerach = [];
//所属单位处理
$value['company'] = empty($aInfo['company']) ? $value['company'] : $aInfo['company'];
if(!empty($value['company'])){
$value['company'] =str_replace('，', ',', $value['company']);
$aCompany = explode(',', $value['company']);
}
$value['company'] = empty($aCompany[0]) ? $value['company'] : trim($aCompany[0],',');
//简介处理
$value['introduction'] = empty($aInfo['introduction']) ? $value['introduction'] : $aInfo['introduction'];
//职称处理
$value['technical'] = empty($aInfo['technical']) ? $value['technical'] : $aInfo['technical'];
$sAuthorInfo .= $value['company'].$value['author_name'].$value['technical'].',';
if(empty($value['introduction'])){//简介为空不处理
$sTechnical = empty($value['technical']) ? '老师' : $value['technical'];
$sAuthorInfo .= $value['company'].$value['author_name'].$sTechnical.',';
continue;
}
//缓缓处理进行模版替换
foreach ($value as $k => $val) {
if($k == 'icon'){//头像拼接处理
$sIconInfo = empty($val) ? $aLogo['url'] : $val;
$aParsedUrl = parse_url($sIconInfo);
if(empty($aParsedUrl['scheme'])){
$aAuthorSerach['{###icon_path###}'] = trim($this->sSubmissionUrl,'/').$this->sUserIcon.$val;
}else{
$aAuthorSerach['{###icon_path###}'] = $sIconInfo;
}
$aAuthorSerach['{###icon_path###}'] = trim($aAuthorSerach['{###icon_path###}'],'/');
}else{
$aAuthorSerach['{###'.$k.'###}'] = trim($val,'/');
}
}
$aAuthorSerachInfo[] = str_replace(array_keys($aAuthorSerach), array_values($aAuthorSerach), $sAuthorTemplate);
$aTranslateAuthorList[$value['email']] = [
'author_name' => empty($value['author_name']) ? '' : $value['author_name'],
'technical' => empty($value['technical']) ? '' : $value['technical'],
'introduction' => empty($value['introduction']) ? '' : $value['introduction'],
'company' => empty($value['company']) ? '' : $value['company']
];
$aAuthorSerachInfo[] = $value;
}
//请求AI翻译
if(!empty($sAuthorInfo)){
$aTranslate = array_merge(['author_info' => trim($sAuthorInfo,',')],$aTranslate);
}
if(!empty($aTranslateAuthorList)){
$aTranslate = array_merge($aTranslateAuthorList,$aTranslate);
}
if(!empty($aTranslate)){
$aSearch['{#content#}'] = json_encode($aTranslate);
$oOpenAi = new OpenAi;
$aTranslate = $oOpenAi->buildTranslatePrompt($aSearch);
$aTranslate = empty($aTranslate['data']) ? [] : $aTranslate['data'];
}
//翻译内容处理
if(!empty($aAuthorSerachInfo)){
foreach ($aAuthorSerachInfo as $key => $value) {
$aInfo = empty($aTranslate[$value['email']]) ? [] : $aTranslate[$value['email']];
$value['author_name'] = empty($aInfo['author_name']) ? $value['author_name'] : $aInfo['author_name'];
$value['technical'] = empty($aInfo['technical']) ? $value['technical'] : $aInfo['technical'];
$value['introduction'] = empty($aInfo['introduction']) ? $value['introduction'] : $aInfo['introduction'];
$value['company'] = empty($aInfo['company']) ? $value['company'] : $aInfo['company'];
$aAuthorSerach = [];
foreach ($value as $k => $val) {
if($k == 'icon'){//头像拼接处理
$sIconInfo = empty($val) ? $aLogo['url'] : $val;
$aParsedUrl = parse_url($sIconInfo);
if(empty($aParsedUrl['scheme'])){
$aAuthorSerach['{###icon_path###}'] = trim($this->sSubmissionUrl,'/').$this->sUserIcon.$val;
}else{
$aAuthorSerach['{###icon_path###}'] = $sIconInfo;
}
$aAuthorSerach['{###icon_path###}'] = trim($aAuthorSerach['{###icon_path###}'],'/');
}else{
$aAuthorSerach['{###'.$k.'###}'] = trim($val,'/');
}
}
$aAuthorSerachInfo[$key] = str_replace(array_keys($aAuthorSerach), array_values($aAuthorSerach), $sAuthorTemplate);
}
}
//通讯作者列表
//获取通讯作者汇总模版
$sAuthorTemplatePath = $sTemplatePath.'/author_summary.html';
@@ -855,7 +770,7 @@ class Aiarticle extends Base
$sAuthorTemplate = empty($sInfo) ? '' : str_replace('{###author_info###}', $sInfo, $sAuthorTemplate);
//通讯作者字符串处理
$sAuthorInfo = trim($sAuthorInfo,',');
$sAuthorInfo = empty($aTranslate['author_info']) ? '' : $aTranslate['author_info'];
return json_encode(['status' => 1,'msg' => 'success','data' => ['author_info' => $sAuthorTemplate,'author_string' => trim($sAuthorInfo,';')]]);
}

15
application/api/controller/Aireview.php
View File

@@ -5,6 +5,7 @@ namespace app\api\controller;
use app\api\controller\Base;
use think\Db;
use app\common\OpenAi;
use app\common\Article;
/**
* @title AI审核文章
* @description 对接OPENAI接口
@@ -53,7 +54,7 @@ class Aireview extends Base
}
//实例化公共方法
$oOpenAi = new OpenAi;
$oArticle = new Article;
if($aArticle['state'] > 4 ){
//查询文章内容
$aWhere['type'] = 0;
@@ -70,7 +71,7 @@ class Aireview extends Base
}
}else{
$aFile = json_decode($oOpenAi->getFileContent(['article_id' => $iArticleId]),true);
$aFile = json_decode($oArticle->getFileContent(['article_id' => $iArticleId]),true);
$aFile = empty($aFile['data']) ? [] : $aFile['data'];
$aArticleMain = empty($aFile['mains']) ? [] : $aFile['mains'];
}
@@ -92,7 +93,7 @@ class Aireview extends Base
$aSearch['{journal_name}'] = '传统医学研究';
}
//查询期刊内容
$aJournalPaperArt = json_decode($oOpenAi->getJournalPaperArt($aJournal),true);
$aJournalPaperArt = json_decode($oArticle->getJournalPaperArt($aJournal),true);
$sJournalContent = empty($aJournalPaperArt['data']) ? '' : implode('', $aJournalPaperArt['data']);
$sJournalContent = empty($sJournalContent) ? $aJournal['scope'] : $sJournalContent;
$aSearch['{scope}'] = $sJournalContent;//期刊范围
@@ -222,12 +223,14 @@ class Aireview extends Base
}
//查询文章
$aArticle = Db::table('t_article')->field('article_id,abstrart,keywords,journal_id,title,state')->where('article_id',$aParam['article_id'])->find();
$aArticle = Db::name('article')->field('article_id,abstrart,keywords,journal_id,title,state')->where('article_id',$aParam['article_id'])->find();
if(empty($aArticle)){
return json_encode(array('status' => 3,'msg' => 'No articles requiring review were found' ));
}
//查询文章审核内容
$aAiReview = Db::table('t_article_ai_review')->where('article_id',$aParam['article_id'])->find();
return json_encode(array('status' => 1,'msg' => 'Successfully obtained article review content','data' => $aAiReview,'article_data' => $aArticle));
$oAireview = new \app\common\Aireview;
$aAireview = $oAireview->get($aParam);
$aAireview = empty($aAireview['data']) ? [] : $aAireview['data'];
return json_encode(array('status' => 1,'msg' => 'Successfully obtained article review content','data' => $aAireview,'article_data' => $aArticle));
}
}

184
application/api/job/ArticleAiCreateContent.php
View File

@@ -1,163 +1,125 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\api\controller\Aiarticle;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\api\controller\Aiarticle;
class ArticleAiCreateContent
{
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/ArticleAiCreateContent_' . 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("-----------队列任务开始-----------");
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
// 获取文章ID
$iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];
if (empty($iArticleId)) {
$this->log("无效的article_id，删除任务");
$this->oQueueJob->log("无效的article_id，删除任务");
$job->delete();
return;
}
// 生成唯一任务标识
$sClassName = get_class($this);
$sRedisKey = "queue_job:{$sClassName}:{$iArticleId}";
$sRedisValue = uniqid() . '_' . getmypid(); // 增加进程ID确保唯一性
$sRedisValue = uniqid() . '_' . getmypid();
$lockExpire = $this->lockExpire;
// 尝试获取Redis锁（原子操作）
$isLocked = $this->queueJob->setRedisLock($sRedisKey, $sRedisValue, 86400);
$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 {
// 执行核心任务
//生成内容
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->create($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([
$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} | 执行日志:{$sMsg}");
} 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();
}
}

124
application/api/job/RecommendArticleField.php Normal file
View 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}｜执行日志:{$sMsg}");
} 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();
}
}
}

172
application/api/job/RecommendReviewer.php
View File

@@ -1,76 +1,106 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\common\Reviewer;
class RecommendReviewer
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/RecommendReviewer_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/RecommendReviewer_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//获取文章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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
// 获取文章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);
}
}
$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 {
//获取推荐审稿人信息
$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'];//推荐数量
@@ -100,33 +130,29 @@ class RecommendReviewer
$sMsg .= empty($aResult['msg']) ? 'Reviewer data insertion failed' : $aResult['msg'];
}
}
//更新任务状态
$aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg];
$oQueueJob->updateLog($aParam);
//删除任务
//更新日志
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

173
application/api/job/RelatedArticle.php
View File

@@ -1,104 +1,129 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\common\JournalArticle;
class RelatedArticle
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/RelatedArticle_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/RelatedArticle_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//文章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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
// 获取文章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);
}
}
$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 {
//查询文章所关联的文章
$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);
//删除任务
//更新日志
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

186
application/api/job/ReviewerScore.php
View File

@@ -1,103 +1,133 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\common\Reviewer;
class ReviewerScore
{
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
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()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/ReviewerScore_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/ReviewerScore_'.date('Ymd').'.log';
$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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
// 获取文章ID
$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'];
if (empty($iArticleId)) {
$this->oQueueJob->log("无效的article_id，删除任务");
$job->delete();
return;
}
// 生成唯一任务标识
$sClassName = get_class($this);
$sRedisKey = "queue_job:{$sClassName}:{$iArticleId}:{$iReviewerId}:{$iArtRevId}";
$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 {
// 执行核心任务
$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);
//删除任务
$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(); // 强制垃圾回收
//更新日志
$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} | 执行日志:{$sMsg}");
} 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();
}
}
}

173
application/api/job/RevisionReviewer.php
View File

@@ -1,100 +1,129 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\common\Reviewer;
class RevisionReviewer
{
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
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()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/RevisionReviewer_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/RevisionReviewer_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//参数
$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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
// 获取文章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);
}
}
$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 {
//获取符合条件的文章审稿人信息
$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);
//删除任务
//更新日志
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

163
application/api/job/SendRelatedArticleEmail.php
View File

@@ -1,41 +1,47 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\common\JournalArticle;
class SendRelatedArticleEmail
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/SendRelatedArticleEmail_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/SendRelatedArticleEmail_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
//文章ID
// 获取文章ID
$iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];
//作者邮箱
$email = empty($data['email']) ? '' : $data['email'];
@@ -57,35 +63,58 @@ class SendRelatedArticleEmail
$journal_id = empty($data['journal_id']) ? '' : $data['journal_id'];
//期刊issn
$journal_issn = empty($data['journal_issn']) ? '' : $data['journal_issn'];
//获取方法名
$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($result != 1){
if (empty($iArticleId) || empty($email)) {
$this->oQueueJob->log("无效的article_id，删除任务");
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
return;
}
// 生成唯一任务标识
$sClassName = get_class($this);
$sRedisKey = "queue_job:{$sClassName}:{$iArticleId}:{$related_article_id}:{$article_author_id}:{$email}";
$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,
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'create_time' => time(),
'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE)
'params' => json_encode($data, self::JSON_OPTIONS)
];
//执行任务
try {
//添加任务日志
$sMsg = '关联文章任务处理成功';
$iLogId = $this->addLog($aParam);
$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 {
//查询是否发送过邮件
$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);
@@ -105,34 +134,30 @@ class SendRelatedArticleEmail
//添加邮件发送日志
$iId = JournalArticle::addLog($aEmailLog);
}
//更新任务状态
$aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg];
$oQueueJob->updateLog($aParam);
//删除任务
//更新日志
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

161
application/api/job/SendReviewEmail.php
View File

@@ -1,40 +1,48 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\common\Reviewer;
class SendReviewEmail
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/SendReviewEmail_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/SendReviewEmail_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
//文章ID
// 获取文章ID
// 获取文章ID
$iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];
//作者邮箱
$email = empty($data['email']) ? '' : $data['email'];
@@ -54,35 +62,59 @@ class SendReviewEmail
$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($result != 1){
if (empty($iArticleId)) {
$this->oQueueJob->log("无效的article_id，删除任务");
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
return;
}
//任务数组
// 生成唯一任务标识
$sClassName = get_class($this);
$sRedisKey = "queue_job:{$sClassName}:{$iArticleId}:{$art_rev_id}:{$reviewer_id}:{$email}";
$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,
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'create_time' => time(),
'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE)
'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 {
//添加任务日志
$sMsg = '发送审稿邀请邮件任务处理成功';
$iLogId = $this->addLog($aParam);
//发送邮件
//查询是否发送过邮件
$oReviewer = new Reviewer;
if($type != 3){
@@ -104,34 +136,29 @@ class SendReviewEmail
//添加邮件发送日志
$iId = $oReviewer->addLog($aEmailLog);
}
//更新日志
$this->oQueueJob->updateLog([
'log_id' => $iLogId,
'status' => 1,
'update_time' => time(),
'error' => $sMsg
]);
//更新任务状态
$aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg];
$oQueueJob->updateLog($aParam);
//删除任务
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

168
application/api/job/WechatDraft.php
View File

@@ -1,102 +1,128 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\api\controller\Aiarticle;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\api\controller\Aiarticle;
class WechatDraft
{
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
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()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/WechatDraft_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/WechatDraft_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//文章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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
}
}
$this->oQueueJob->flushLog();
return;
}
//任务数组
$aParam = [
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'create_time' => time(),
'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE)
'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 {
//添加任务日志
$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);
//更新日志
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

170
application/api/job/WechatDraftPublish.php
View File

@@ -1,104 +1,128 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\api\controller\Aiarticle;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\api\controller\Aiarticle;
class WechatDraftPublish
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/WechatDraftPublish_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/WechatDraftPublish_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//文章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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
}
}
$this->oQueueJob->flushLog();
return;
}
//任务数组
$aParam = [
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'create_time' => time(),
'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE)
'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 {
//添加任务日志
$sMsg = '草稿箱发布任务处理成功';
$iLogId = $this->addLog($aParam);
//发布草稿箱
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->publishDraft($data),true);
$iStatus = empty($aResult['status']) ? 0 : $aResult['status'];
$aResult = json_decode($oAiarticle->publishDraft($data),true);
$sMsg = empty($aResult['msg']) ? '草稿箱发布失败' : $aResult['msg'];
//更新任务状态
$aParam = ['log_id' => $iLogId,'status' => 1,'update_time' => time(),'error' => $sMsg];
$oQueueJob->updateLog($aParam);
//删除任务
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

164
application/api/job/WechatMaterial.php
View File

@@ -1,102 +1,128 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\api\controller\Aiarticle;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\api\controller\Aiarticle;
class WechatMaterial
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/WechatMaterial_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/WechatMaterial_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//文章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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
}
}
$this->oQueueJob->flushLog();
return;
}
//任务数组
$aParam = [
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'create_time' => time(),
'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE)
'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 {
//添加任务日志
$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);
//删除任务
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

165
application/api/job/WechatQueryStatus.php
View File

@@ -1,103 +1,128 @@
<?php
namespace app\api\job;
use think\queue\Job;
use app\api\controller\Aiarticle;
use app\common\QueueJob;
use app\common\QueueRedis;
use app\api\controller\Aiarticle;
class WechatQueryStatus
{
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;
// 最多重试1次
public $tries = 1;
// 任务日志添加
public function addLog($aParam = [])
public function __construct()
{
//实例化
$oQueueJob = new QueueJob;
$iLogId = $oQueueJob->addLog($aParam);
return $iLogId;
}
// 任务日志修改
public function updateLog($aParam = [])
{
//实例化
$oQueueJob = new QueueJob;
return $oQueueJob->updateLog($aParam);
$this->logPath = ROOT_PATH . 'public/queue_log/WechatQueryStatus_' . 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("-----------队列任务开始-----------");
$sLogPath = ROOT_PATH.'public/queue_log/WechatQueryStatus_'.date('Ymd').'.log';
$sTime = date('H:i:s');
file_put_contents($sLogPath,'-----------Queue job started:'.$sTime.'-----------');
//文章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($result != 1){
$job->delete();
file_put_contents($sLogPath,'-----------Queue job already:'.$result."===".$sRedisKey.'==='.$iArticleId."===".$sTime.'-----------');
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
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);
}
}
$this->oQueueJob->flushLog();
return;
}
//任务数组
$aParam = [
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'job_id' => $sRedisKey,
'job_class' => $sClassName,
'status' => 0,
'create_time' => time(),
'params' => empty($data) ? '暂无参数' : json_encode($data, JSON_UNESCAPED_UNICODE)
'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 {
//添加任务日志
$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);
//删除任务
$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} | 执行日志:{$sMsg}");
file_put_contents($sLogPath,'-----------Queue job end:'.$sTime.'-----------');
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
} 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(); // 强制垃圾回收
$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();
}
}
}

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} | 执行日志:{$sMsg}");
} 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();
}
}