公共方法和任务新增

2025-08-15 14:40:02 +08:00
parent 65a799067c
commit 41007a1302
5 changed files with 722 additions and 0 deletions
--- a/application/api/job/ArticleReview.php
+++ b/application/api/job/ArticleReview.php
@@ -0,0 +1,78 @@
+<?php
+namespace app\api\job;
+
+use think\queue\Job;
+use app\common\QueueJob;
+use app\common\QueueRedis;
+use app\api\controller\Aireview;
+class ArticleReview
+{
+    private $oQueueJob;
+    private $QueueRedis;
+    private $completedExprie = 3600;
+
+    public function __construct()
+    {
+        $this->oQueueJob = new QueueJob;
+        $this->QueueRedis = QueueRedis::getInstance();
+    }
+
+    public function fire(Job $job, $data)
+    {
+        //任务开始判断
+        $this->oQueueJob->init($job);
+
+        // 获取 Redis 任务的原始数据
+        $rawBody = empty($job->getRawBody()) ? '' : $job->getRawBody();
+        $jobData = empty($rawBody) ? [] : json_decode($rawBody, true);
+        $jobId = empty($jobData['id']) ? 'unknown' : $jobData['id'];
+
+        $this->oQueueJob->log("-----------队列任务开始-----------");
+        $this->oQueueJob->log("当前任务ID: {$jobId}, 尝试次数: {$job->attempts()}");
+
+        // 获取文章ID
+        $iArticleId = empty($data['article_id']) ? 0 : $data['article_id'];
+        if (empty($iArticleId)) {
+            $this->oQueueJob->log("无效的article_id，删除任务");
+            $job->delete();
+            return;
+        }
+        try {
+            
+            // 生成Redis键并尝试获取锁
+            $sClassName = get_class($this);
+            $sRedisKey = "queue_job:{$sClassName}:{$iArticleId}";
+            $sRedisValue = uniqid() . '_' . getmypid();
+            if (!$this->oQueueJob->acquireLock($sRedisKey, $sRedisValue, $job)) {
+                return; // 未获取到锁，已处理
+            }
+
+            //生成内容
+            $oAireview = new Aireview;
+            $response = $oAireview->review($data);
+            // 验证API响应
+            if (empty($response)) {
+                throw new RuntimeException("OpenAI API返回空结果");
+            }
+            // 检查JSON解析错误
+            $aResult = json_decode($response, true);
+            if (json_last_error() !== JSON_ERROR_NONE) {
+                throw new RuntimeException("解析OpenAI响应失败: " . json_last_error_msg() . " | 原始响应: {$response}");
+            }
+            $sMsg = empty($aResult['msg']) ? 'success' : $aResult['msg'];
+            //更新完成标识
+            $this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
+            $job->delete();
+            $this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
+
+        } catch (\RuntimeException $e) {
+            $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } catch (\LogicException $e) {
+            $this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } catch (\Exception $e) {
+            $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } finally {
+            $this->oQueueJob->finnal();
+        }
+    }
+}
--- a/application/api/job/ArticleReviewForQueue.php
+++ b/application/api/job/ArticleReviewForQueue.php
@@ -0,0 +1,81 @@
+<?php
+namespace app\api\job;
+
+use think\queue\Job;
+use app\common\OpenAi;
+use app\common\QueueJob;
+use app\common\QueueRedis;
+
+class ArticleReviewForQueue
+{
+    private $oQueueJob;
+    private $QueueRedis;
+    private $completedExprie = 3600;  // 完成状态过期时间
+    public function __construct()
+    {
+        $this->oQueueJob = new QueueJob;
+        $this->QueueRedis = QueueRedis::getInstance();
+    }
+
+    public function fire(Job $job, $data)
+    {
+        //任务开始判断
+        $this->oQueueJob->init($job);
+
+        // 获取 Redis 任务的原始数据
+        $rawBody = empty($job->getRawBody()) ? '' : $job->getRawBody();
+        $jobData = empty($rawBody) ? [] : json_decode($rawBody, true);
+        $jobId = empty($jobData['id']) ? 'unknown' : $jobData['id'];
+
+        $this->oQueueJob->log("-----------队列任务开始-----------");
+        $this->oQueueJob->log("当前任务ID: {$jobId}, 尝试次数: {$job->attempts()}");
+
+        try {
+           
+            // 验证任务数据完整性
+            $iRedisId = empty($data['article_id']) ? 0 : $data['article_id'];
+            $sChunkIndex = empty($data['chunkIndex']) ? 0 : $data['chunkIndex'];
+            if (empty($iRedisId)) {
+                $this->oQueueJob->log("无效的redis_id，删除任务");
+                $job->delete();
+                return;
+            }
+
+            // 生成Redis键并尝试获取锁
+            $sClassName = get_class($this);
+            $sRedisKey = "queue_job:{$sClassName}:{$iRedisId}:{$sChunkIndex}";
+            $sRedisValue = uniqid() . '_' . getmypid();
+            if (!$this->oQueueJob->acquireLock($sRedisKey, $sRedisValue, $job)) {
+                return; // 未获取到锁，已处理
+            }
+
+            // 执行核心任务
+            $oOpenAi = new OpenAi;
+            $response = $oOpenAi->articleReviewForQueue($data);
+            // 验证API响应
+            if (empty($response)) {
+                throw new RuntimeException("OpenAI API返回空结果");
+            }
+            // 检查JSON解析错误
+            $aResult = json_decode($response, true);
+            if (json_last_error() !== JSON_ERROR_NONE) {
+                throw new RuntimeException("解析OpenAI响应失败: " . json_last_error_msg() . " | 原始响应: {$response}");
+            }
+            $sMsg = empty($aResult['msg']) ? '内容生成成功' : $aResult['msg'];
+
+            // 更新完成标识
+            $this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie, $sRedisValue);
+            $job->delete();
+            $this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
+
+        } catch (RuntimeException $e) {
+            $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } catch (LogicException $e) {
+            $this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } catch (Exception $e) {
+            $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } finally {
+            $this->oQueueJob->finnal();
+        }
+    }
+}
--- a/application/api/job/ArticleReviewForQueueChunk.php
+++ b/application/api/job/ArticleReviewForQueueChunk.php
@@ -0,0 +1,81 @@
+<?php
+namespace app\api\job;
+
+use think\queue\Job;
+use app\common\OpenAi;
+use app\common\QueueJob;
+use app\common\QueueRedis;
+
+class ArticleReviewForQueueChunk
+{
+    private $oQueueJob;
+    private $QueueRedis;
+    private $completedExprie = 3600;  // 完成状态过期时间
+    public function __construct()
+    {
+        $this->oQueueJob = new QueueJob;
+        $this->QueueRedis = QueueRedis::getInstance();
+    }
+
+    public function fire(Job $job, $data)
+    {
+        //任务开始判断
+        $this->oQueueJob->init($job);
+
+        // 获取 Redis 任务的原始数据
+        $rawBody = empty($job->getRawBody()) ? '' : $job->getRawBody();
+        $jobData = empty($rawBody) ? [] : json_decode($rawBody, true);
+        $jobId = empty($jobData['id']) ? 'unknown' : $jobData['id'];
+
+        $this->oQueueJob->log("-----------队列任务开始-----------");
+        $this->oQueueJob->log("当前任务ID: {$jobId}, 尝试次数: {$job->attempts()}");
+
+        try {
+           
+            // 验证任务数据完整性
+            $iRedisId = empty($data['article_id']) ? 0 : $data['article_id'];
+            $sChunkIndex = empty($data['chunkIndex']) ? 0 : $data['chunkIndex'];
+            if (empty($iRedisId)) {
+                $this->oQueueJob->log("无效的redis_id，删除任务");
+                $job->delete();
+                return;
+            }
+
+            // 生成Redis键并尝试获取锁
+            $sClassName = get_class($this);
+            $sRedisKey = "queue_job:{$sClassName}:{$iRedisId}:{$sChunkIndex}";
+            $sRedisValue = uniqid() . '_' . getmypid();
+            if (!$this->oQueueJob->acquireLock($sRedisKey, $sRedisValue, $job)) {
+                return; // 未获取到锁，已处理
+            }
+
+            // 执行核心任务
+            $oOpenAi = new OpenAi;
+            $response = $oOpenAi->articleReviewForQueueChunk($data);
+            // 验证API响应
+            if (empty($response)) {
+                throw new RuntimeException("OpenAI API返回空结果");
+            }
+            // 检查JSON解析错误
+            $aResult = json_decode($response, true);
+            if (json_last_error() !== JSON_ERROR_NONE) {
+                throw new RuntimeException("解析OpenAI响应失败: " . json_last_error_msg() . " | 原始响应: {$response}");
+            }
+            $sMsg = empty($aResult['msg']) ? '内容生成成功' : $aResult['msg'];
+
+            // 更新完成标识
+            $this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie, $sRedisValue);
+            $job->delete();
+            $this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
+
+        } catch (RuntimeException $e) {
+            $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } catch (LogicException $e) {
+            $this->oQueueJob->handleNonRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } catch (Exception $e) {
+            $this->oQueueJob->handleRetryableException($e,$sRedisKey,$sRedisValue, $job);
+        } finally {
+            $this->oQueueJob->finnal();
+        }
+    }
+}
--- a/application/api/job/TranslateContentForQueue.php
+++ b/application/api/job/TranslateContentForQueue.php
@@ -0,0 +1,102 @@
+<?php
+namespace app\api\job;
+
+use think\queue\Job;
+use app\common\OpenAi;
+use app\common\QueueJob;
+use app\common\QueueRedis;
+
+class TranslateContentForQueue
+{
+    private $oQueueJob;
+    private $QueueRedis;
+    private $maxRetries = 2;
+    private $lockExpire = 1800;
+    private $completedExprie = 3600;
+    const JSON_OPTIONS = JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR;
+
+    public function __construct()
+    {
+        $this->oQueueJob = new QueueJob;
+        $this->QueueRedis = QueueRedis::getInstance();
+    }
+
+    public function fire(Job $job, $data)
+    {
+        $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);
+            return;
+        }
+
+        $iRedisId = empty($data['article_id']) ? 0 : $data['article_id'];
+        $sChunkIndex = empty($data['chunkIndex']) ? 0 : $data['chunkIndex'];
+        if (empty($iRedisId)) {
+            $this->oQueueJob->log("无效的article_id，删除任务");
+            $job->delete();
+            return;
+        }
+
+        $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) {
+            $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);
+                }
+            }
+            return;
+        }
+
+        try {
+            // 执行核心任务前再次检查连接
+            $result = $this->oQueueJob->checkDbConnection();
+            if (!$result) {
+                throw new \RuntimeException("数据库连接异常，无法执行核心任务");
+            }
+            $oOpenAi = new OpenAi;
+            $aResult = json_decode($oOpenAi->translateContentForQueue($data), true);
+            $sMsg = empty($aResult['msg']) ? '内容翻译成功' : $aResult['msg'];
+            //更新完成标识
+            $this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie,$sRedisValue);
+            $job->delete();
+            $this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | 执行日志:{$sMsg}");
+
+        } catch (\RuntimeException $e) {
+            $this->oQueueJob->handleRetryableException($e, $sRedisKey,$sRedisValue,$job);
+        } catch (\LogicException $e) {
+            $this->oQueueJob->handleNonRetryableException($e, $sRedisKey,$sRedisValue,$job);
+        } catch (\Exception $e) {
+            $this->oQueueJob->handleRetryableException($e, $sRedisKey,$sRedisValue,$job);
+        } finally {
+            $executionTime = microtime(true) - $startTime;
+            $this->oQueueJob->log("任务执行完成，耗时: " . number_format($executionTime, 4) . "秒");
+            gc_collect_cycles();
+        }
+    }
+}