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-30 14:35:24 +08:00
parent 8358a9ed5f 17cf1c7ddd
commit 7fb2cfa30e
21 changed files with 727 additions and 874 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -85,10 +85,8 @@ class Aiarticle extends Base
}
//查询AI内容是否生成
$aAiArticle = json_decode($this->getAiArticle(['article_id' => $iArticleId]),true);
$aAiArticleContent = empty($aAiArticle['data']) ? [] : $aAiArticle['data'];
$aAiArticle = empty($aAiArticleContent['ai_article']) ? [] : $aAiArticleContent['ai_article'];
$aWhere = ['is_delete' => 2,'article_id' => $iArticleId];
$aAiArticle = Db::name('ai_article')->where($aWhere)->find();
$iId = empty($aAiArticle['ai_article_id']) ? 0 : $aAiArticle['ai_article_id'];
if(empty($aAiArticle)){
//插入t_ai_article数据
@@ -100,6 +98,7 @@ class Aiarticle extends Base
}
$aAiArticle = array_merge(['ai_article_id' => $iId,'article_id' => $iArticleId,'is_generate' => 2],$aInsert);
}
//判断是否生成
if(!empty($aAiArticle['is_generate']) && $aAiArticle['is_generate'] == 1){
return json_encode(['status' => 5,'msg' => 'The data has been generated, please proceed with the next steps']);

57
application/api/controller/Article.php
View File

@@ -2005,6 +2005,21 @@ class Article extends Base
return jsonError("The article can only be added in state with editor at least");
}
//获取审稿人最后一次邀请时间 chengxiaoling 20250724 start
$iUserId = empty($data['uid']) ? 0 : $data['uid'];
if(empty($iUserId)){
return jsonError("Please select the reviewers to invite!");
}
//判断距离上次邀请审稿是否超过10天
$aWhere = ['reviewer_id' => $iUserId];
$iTeenDaysLater = strtotime('-10 days');// 计算10天之前的时间戳
$aUserInfo = Db::name('user_reviewer_info')->field('last_invite_time')->where($aWhere)->where('t_user_reviewer_info.last_invite_time', '<', $iTeenDaysLater)
->whereOr('t_user_reviewer_info.last_invite_time', '=', 0)->find();
if(empty($aUserInfo)){
return jsonError("The time since the last invitation for review by the reviewer has not exceeded 10 days!");
}
//获取审稿人最后一次邀请时间 chengxiaoling 20250724 end
//增加信息到文章审稿表
$insert_data['reviewer_id'] = $data['uid'];
$insert_data['article_id'] = $data['articleId'];
@@ -2013,6 +2028,14 @@ class Article extends Base
$insert_data['state'] = 5;
$res = $this->article_reviewer_obj->insertGetId($insert_data);
//更新审稿人最后一次审稿时间 chengxiaoling 20250724 start
if(!empty($res) && !empty($insert_data['reviewer_id'])){
$aUpdate = ['last_invite_time'=>time()];
$aWhere = ['reviewer_id' => $iUserId];
$updateResult = Db::name('user_reviewer_info')->where($aWhere)->limit(1)->update($aUpdate);
}
//更新审稿人最后一次审稿时间 chengxiaoling 20250724 end
//修改文章状态->审稿中
$this->article_obj->where('article_id', $data['articleId'])->update(['state' => 2]);
@@ -3063,7 +3086,7 @@ class Article extends Base
'journal' => 'require',
'title' => 'require',
'type' => 'require',
'major' => 'require',
// 'major' => 'require',
'abstrart' => 'require'
]);
if (!$rule->check($data)) {
@@ -3114,6 +3137,12 @@ class Article extends Base
$inset_data['ctime'] = time();
$inset_data['state'] = -1;
//新增字段是否使用AI及使用说明 chengxiaoling 20250725 start
$inset_data['is_use_ai'] = empty($data['is_use_ai']) ? 2 : $data['is_use_ai']; //是否使用AI1是2否
$inset_data['use_ai_explain'] = isset($data['use_ai_explain']) ? $data['use_ai_explain'] : '';
//新增字段是否使用AI及使用说明 chengxiaoling 20250725 end
$article_id = $this->article_obj->insertGetId($inset_data);
} else {
@@ -3141,9 +3170,20 @@ class Article extends Base
$up["approval"] = 0;
$up['approval_content'] = isset($data["approval_content"]) ? $data["approval_content"] : '';//trim($data['approval_content']);
}
//新增字段是否使用AI及使用说明 chengxiaoling 20250725 start
$up['is_use_ai'] = empty($data['is_use_ai']) ? 2 : $data['is_use_ai']; //是否使用AI1是2否
$up['use_ai_explain'] = isset($data['use_ai_explain']) ? $data['use_ai_explain'] : '';//使用AI说明
//新增字段是否使用AI及使用说明 chengxiaoling 20250725 end
$this->article_obj->where('article_id', $article_id)->update($up);
}
changeArticleMajor($article_id, $data['major']);
//注释文章筛选领域添加修改为AI推荐领域，在第四步可以查看修改 chengxiaoling 20250722
// changeArticleMajor($article_id,$data['major']);
if(!empty($article_id)){//AI推荐领域队列执行
$sQueueId = \think\Queue::push('app\api\job\RecommendArticleField@fire',['article_id' => $article_id], 'RecommendArticleField');
}
//注释文章筛选领域添加修改为AI推荐领域，在第四步可以查看修改 chengxiaoling 20250722
return jsonSuccess(['article_id' => $article_id]);
}
@@ -3451,6 +3491,15 @@ class Article extends Base
}
$this->article_obj->where('article_id', $data['article_id'])->update($update_l);
$this->ai_scor($data['article_id']);
//判断是否有文章领域 进行更新操作 chengxiaoling 20250722 start
$sMajorData = empty($data['article_field']) ? '' : $data['article_field'];//文章领域
$iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];//文章ID
if(!empty($sMajorData) && !empty($iArticleId)){
$this->updateArticleField(['article_id' => $iArticleId,'article_field' => $sMajorData]);
}
//判断是否有文章领域 进行更新操作 chengxiaoling 20250722 end
return json(['code' => 0]);
}
@@ -4617,7 +4666,7 @@ class Article extends Base
public function getArticleField($aParam = [])
{
$aParam = empty($aParam) ? $this->request->post() : $this->request->post();
$aParam = empty($aParam) ? $this->request->post() : $aParam;
if (empty($aParam['article_id'])) {
return json_encode(['status' => 2, 'msg' => 'Please select a Article']);
@@ -4646,7 +4695,7 @@ class Article extends Base
public function updateArticleField($aParam = [])
{
$aParam = empty($aParam) ? $this->request->post() : $this->request->post();
$aParam = empty($aParam) ? $this->request->post() : $aParam;
$iArticleId = empty($aParam['article_id']) ? 0 : $aParam['article_id'];
if (empty($iArticleId)) {
return json_encode(['status' => 2, 'msg' => 'Please select a article']);

39
application/api/controller/Queueinfo.php
View File

@@ -33,7 +33,7 @@ class Queueinfo extends Base
$redis->select($config['select']);
// 获取不同队列类型的任务数
$aQueue = empty($aParam['queue_name']) ? ['ArticleAiCreateContent','RecommendReviewer','RelatedArticle','ReviewerScore','RevisionReviewer','SendRelatedArticleEmail','SendReviewEmail','WechatDraft','WechatDraftPublish','WechatMaterial','WechatQueryStatus'] : [$aParam['queue_name']];
$aQueue = empty($aParam['queue_name']) ? ['ArticleAiCreateContent','RecommendReviewer','RelatedArticle','ReviewerScore','RevisionReviewer','SendRelatedArticleEmail','SendReviewEmail','WechatDraft','WechatDraftPublish','WechatMaterial','WechatQueryStatus','createFieldForQueue','RecommendArticleField'] : [$aParam['queue_name']];
foreach ($aQueue as $key => $value) {
$types[$value] = [
'pending' => 'queues:'.$value,
@@ -53,7 +53,8 @@ class Queueinfo extends Base
}
}
// 关闭 Redis 连接（关键步骤）
$redis->close();
echo '<pre>';var_dump($counts);
}
public function removeKey() {
@@ -99,7 +100,8 @@ echo '<pre>';var_dump($counts);
}
}
// 关闭 Redis 连接（关键步骤）
$redis->close();
echo '<pre>';var_dump($counts);
}
// Redis值删除
@@ -142,7 +144,8 @@ echo '<pre>';var_dump($counts);
}
echo '<pre>';var_dump($sQueueInfo,$sKey,$iNum);exit;
// 关闭 Redis 连接（关键步骤）
$redis->close();
}
@@ -171,36 +174,56 @@ echo '<pre>';var_dump($sQueueInfo,$sKey,$iNum);exit;
$sKey = $redis->get($aParam['key_name']);
var_dump($sKey);
// 关闭 Redis 连接（关键步骤）
$redis->close();
}
public function removeKey1(){
// 获取请求参数
$aParam = $this->request->post();
if(empty($aParam['key_name'])){
if (empty($aParam['key_name'])) {
exit('非法操作');
}
// 获取队列配置（Redis 配置）
$config = \think\Config::get('queue');
// 队列名称默认值处理
$sQueueName = empty($aParam['queue_name']) ? 'ArticleAiCreateContent' : $aParam['queue_name'];
// 初始化 Redis 连接
$redis = new \Redis();
// 连接 Redis 服务器
$redis->connect($config['host'], $config['port']);
// 若有密码则认证
if (!empty($config['password'])) {
$redis->auth($config['password']);
}
// 选择数据库（默认 0）
$redis->select($config['select']);
$sKey = $redis->hGetAll($aParam['key_name']);
// 操作示例：获取哈希表数据
echo '我是HgetAll';
$sKey = $redis->hGetAll($aParam['key_name']);
var_dump($sKey);
// 删除哈希表
$result = $redis->del($aParam['key_name']);
var_dump($result);
// 尝试获取字符串类型的键（验证是否删除）
echo '我是get';
$sKey = $redis->get($aParam['key_name']);
var_dump($sKey);
// 再次删除（确保删除）
$result = $redis->del($aParam['key_name']);
var_dump($result);
// 关闭 Redis 连接（关键步骤）
$redis->close();
}
}

16
application/api/controller/Reviewer.php
View File

@@ -2182,11 +2182,19 @@ class Reviewer extends Base
if (isset($data['major_id'])&&$data['major_id']!=0){
$where['t_user_reviewer_info.major'] = ['in',$this->majorids($data['major_id'])];
}
// 计算10天之后的时间戳（10天 = 10 * 24 * 60 * 60秒）
$iTeenDaysLater = strtotime('-10 days');
//获取邀请时间超过10天的 chengxiaoling 20250728 start
//获取总条数
$count = $this->reviewer_to_journal_obj
->join("t_user", "t_user.user_id = t_reviewer_to_journal.reviewer_id", "left")
->join("t_user_reviewer_info", "t_user_reviewer_info.reviewer_id = t_reviewer_to_journal.reviewer_id", "left")
->where($where)
->where($where)->where(function($query) use ($iTeenDaysLater) {
$query->where('t_user_reviewer_info.last_invite_time', '<', $iTeenDaysLater)
->whereOr('t_user_reviewer_info.last_invite_time', '=', 0);
})
->count();
if(empty($count)){
return jsonSuccess(['reviewers' => [],'count' => 0]);
@@ -2197,10 +2205,14 @@ class Reviewer extends Base
->join("t_user", "t_user.user_id = t_reviewer_to_journal.reviewer_id", "left")
->join("t_user_reviewer_info", "t_user_reviewer_info.reviewer_id = t_reviewer_to_journal.reviewer_id", "left")
->field('t_user.account,t_user.email,t_user.realname,t_user_reviewer_info.company,t_user_reviewer_info.field,t_user.user_id,t_user.rs_num')
->where($where)
->where($where)->where(function($query) use ($iTeenDaysLater) {
$query->where('t_user_reviewer_info.last_invite_time', '<', $iTeenDaysLater)
->whereOr('t_user_reviewer_info.last_invite_time', '=', 0);
})
->order('t_user.rs_num desc')
->limit($limit_start, $data['pageSize'])
->select();
//获取邀请时间超过10天的 chengxiaoling 20250728 end
if(!empty($list)){
$aUserId = array_column($list, 'user_id');
$aWhere = ['state' => 0,'reviewer_id' => ['in',$aUserId]];

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

@@ -7,25 +7,17 @@ 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;
public function __construct()
{
$this->logPath = ROOT_PATH . 'public/queue_log/ArticleAiCreateContent_' . 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)
@@ -33,11 +25,17 @@ class ArticleAiCreateContent
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,54 +70,32 @@ class ArticleAiCreateContent
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//生成内容
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->create($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);
//更新完成标识
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue, $job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

60
application/api/job/RecommendArticleField.php
View File

@@ -7,25 +7,17 @@ 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)
@@ -33,11 +25,17 @@ class RecommendArticleField
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -45,7 +43,6 @@ class RecommendArticleField
if (empty($iArticleId)) {
$this->oQueueJob->log("无效的article_id，删除任务");
$job->delete();
$this->oQueueJob->flushLog();
return;
}
@@ -72,52 +69,31 @@ class RecommendArticleField
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
$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);
//更新完成标识
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId}｜执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey}｜执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e, $sRedisKey, $sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e, $sRedisKey, $sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e, $sRedisKey, $sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class RecommendReviewer
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/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)
@@ -33,11 +26,17 @@ class RecommendReviewer
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,29 +71,16 @@ class RecommendReviewer
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//获取推荐审稿人信息
$aParam = ['article_id' => $iArticleId,'page' => 1,'size' => empty($data['size']) ? 5 : $data['size']];
$oReviewer = new Reviewer;
@@ -130,28 +116,20 @@ class RecommendReviewer
$sMsg .= empty($aResult['msg']) ? 'Reviewer data insertion failed' : $aResult['msg'];
}
}
//更新日志
$this->oQueueJob->updateLog([
'log_id' => $iLogId,
'status' => 1,
'update_time' => time(),
'error' => $sMsg
]);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie);
//更新完成标识
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class RelatedArticle
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/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)
@@ -33,11 +26,17 @@ class RelatedArticle
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,29 +71,16 @@ class RelatedArticle
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//查询文章所关联的文章
$oJournalArticle = new JournalArticle;
$aResult = json_decode(JournalArticle::get($data),true);
@@ -102,27 +88,19 @@ class RelatedArticle
$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);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class ReviewerScore
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/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)
@@ -33,11 +26,17 @@ class ReviewerScore
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -76,57 +75,34 @@ class ReviewerScore
$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 {
// 执行核心任务
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
$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'];
//更新日志
$this->oQueueJob->updateLog([
'log_id' => $iLogId,
'status' => 1,
'update_time' => time(),
'error' => $sMsg
]);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class RevisionReviewer
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/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)
@@ -33,11 +26,17 @@ class RevisionReviewer
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,29 +71,15 @@ class RevisionReviewer
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//获取符合条件的文章审稿人信息
$aParam = ['article_id' => $iArticleId];
$oReviewer = new Reviewer;
@@ -102,27 +87,19 @@ class RevisionReviewer
$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);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class SendRelatedArticleEmail
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/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)
@@ -33,11 +26,17 @@ class SendRelatedArticleEmail
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -92,29 +91,15 @@ class SendRelatedArticleEmail
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//查询是否发送过邮件
$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);
@@ -136,27 +121,19 @@ class SendRelatedArticleEmail
}
//更新日志
$this->oQueueJob->updateLog([
'log_id' => $iLogId,
'status' => 1,
'update_time' => time(),
'error' => $sMsg
]);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue, $job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class SendReviewEmail
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/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)
@@ -33,16 +26,21 @@ class SendReviewEmail
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
// 获取文章ID
// 获取文章ID
$iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];
//作者邮箱
$email = empty($data['email']) ? '' : $data['email'];
@@ -62,8 +60,8 @@ class SendReviewEmail
$reviewer_id = empty($data['reviewer_id']) ? 0 : $data['reviewer_id'];
//邮件类型
$type = empty($data['type']) ? 1 : $data['type'];
if (empty($iArticleId)) {
$this->oQueueJob->log("无效的article_id，删除任务");
if (empty($iArticleId) || empty($email)) {
$this->oQueueJob->log("无效的article_id/email，删除任务");
$job->delete();
return;
}
@@ -92,29 +90,15 @@ class SendReviewEmail
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//查询是否发送过邮件
$oReviewer = new Reviewer;
if($type != 3){
@@ -137,27 +121,19 @@ class SendReviewEmail
$iId = $oReviewer->addLog($aEmailLog);
}
//更新日志
$this->oQueueJob->updateLog([
'log_id' => $iLogId,
'status' => 1,
'update_time' => time(),
'error' => $sMsg
]);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class WechatDraft
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/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)
@@ -33,11 +26,17 @@ class WechatDraft
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,56 +71,35 @@ class WechatDraft
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//上传草稿箱
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->syncWechat($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);
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class WechatDraftPublish
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/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)
@@ -33,11 +26,17 @@ class WechatDraftPublish
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,56 +71,35 @@ class WechatDraftPublish
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//发布草稿箱
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->publishDraft($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);
//更新日志
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class WechatMaterial
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/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)
@@ -33,11 +26,17 @@ class WechatMaterial
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,56 +71,36 @@ class WechatMaterial
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
//上传素材
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->uploadMaterial($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);
//更新日志
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -11,21 +11,14 @@ class WechatQueryStatus
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/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)
@@ -33,11 +26,17 @@ class WechatQueryStatus
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -72,56 +71,37 @@ class WechatQueryStatus
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
// 查询状态
$oAiarticle = new Aiarticle;
$aResult = json_decode($oAiarticle->queryStatus($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);
//更新日志
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisKey,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisKey,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}

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

@@ -8,25 +8,17 @@ 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;
public function __construct()
{
$this->logPath = ROOT_PATH . 'public/queue_log/createFieldForQueue_' . 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)
@@ -34,11 +26,16 @@ class createFieldForQueue
$startTime = microtime(true);
$this->oQueueJob->log("-----------队列任务开始-----------");
// 检查数据库连接
if (!$this->oQueueJob->checkDbConnection(true)) {
$this->oQueueJob->log("数据库连接失败，无法执行任务，10秒后重试");
$job->release(10);
return;
}
// 检查Redis连接状态
if (!$this->QueueRedis->getConnectionStatus()) {
$this->oQueueJob->log("Redis连接失败，10秒后重试");
$job->release(10);
$this->oQueueJob->flushLog();
return;
}
@@ -47,7 +44,6 @@ class createFieldForQueue
if (empty($iRedisId)) {
$this->oQueueJob->log("无效的redis_id，删除任务");
$job->delete();
$this->oQueueJob->flushLog();
return;
}
@@ -74,53 +70,32 @@ class createFieldForQueue
$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 {
// 执行核心任务前再次检查连接
$result = $this->oQueueJob->checkDbConnection();
if (!$result) {
throw new \RuntimeException("数据库连接异常，无法执行核心任务");
}
$oOpenAi = new OpenAi;
$aResult = json_decode($oOpenAi->createFieldForQueue($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);
//更新完成标识
$this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
$job->delete();
$this->oQueueJob->log("任务执行成功 | 日志ID: {$iLogId} | 执行日志:{$sMsg}");
$this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
} catch (\RuntimeException $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e, $sRedisKey,$sRedisValue,$job);
} catch (\LogicException $e) {
$this->oQueueJob->handleNonRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleNonRetryableException($e, $sRedisKey,$sRedisValue,$job);
} catch (\Exception $e) {
$this->oQueueJob->handleRetryableException($e, $iLogId, $sRedisKey, $job);
$this->oQueueJob->handleRetryableException($e, $sRedisKey,$sRedisValue,$job);
} finally {
$executionTime = microtime(true) - $startTime;
$this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
$this->oQueueJob->flushLog();
gc_collect_cycles();
}
}