diff --git a/application/api/controller/Production.php b/application/api/controller/Production.php index f6f1a0c5..d1e451c7 100644 --- a/application/api/controller/Production.php +++ b/application/api/controller/Production.php @@ -1932,7 +1932,8 @@ class Production extends Base } $this->referToDoi($data['p_article_id']); - (new ReferenceDispatchService())->dispatchRefersByType($data['p_article_id']); + // 每条文献单独入队,接口立即返回;多 worker 可并行处理 200+ 条 + (new ReferenceDispatchService())->enqueueRefersByType($data['p_article_id']); return jsonSuccess([]); } @@ -1999,7 +2000,7 @@ class Production extends Base public function doiTofrag($p_article_id) { - (new ReferenceDispatchService())->dispatchRefersByType($p_article_id); + (new ReferenceDispatchService())->enqueueRefersByType($p_article_id); return jsonSuccess([]); } diff --git a/application/api/job/ReferenceDispatchQueue.php b/application/api/job/ReferenceDispatchQueue.php new file mode 100644 index 00000000..2c1f39bc --- /dev/null +++ b/application/api/job/ReferenceDispatchQueue.php @@ -0,0 +1,70 @@ +oQueueJob = new QueueJob; + $this->QueueRedis = QueueRedis::getInstance(); + } + + public function fire(Job $job, $data) + { + $this->oQueueJob->init($job); + + $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()}"); + + $iPArticleId = empty($data['p_article_id']) ? 0 : intval($data['p_article_id']); + $iPReferId = empty($data['p_refer_id']) ? 0 : intval($data['p_refer_id']); + if (empty($iPArticleId) || empty($iPReferId)) { + $this->oQueueJob->log("无效的 p_article_id 或 p_refer_id,删除任务"); + $job->delete(); + return; + } + + try { + $sClassName = get_class($this); + $sRedisKey = "queue_job:{$sClassName}:{$iPArticleId}:{$iPReferId}"; + $sRedisValue = uniqid() . '_' . getmypid(); + if (!$this->oQueueJob->acquireLock($sRedisKey, $sRedisValue, $job)) { + return; + } + + (new ReferenceDispatchService())->dispatchReferByType($iPReferId); + + $this->QueueRedis->finishJob($sRedisKey, 'completed', $this->completedExprie, $sRedisValue); + $job->delete(); + $this->oQueueJob->log("任务执行成功 | 日志ID: {$sRedisKey} | p_refer_id:{$iPReferId}"); + } 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(); + } + } +} diff --git a/application/common/ReferenceDispatchService.php b/application/common/ReferenceDispatchService.php index 8358f35a..e2ef01d8 100644 --- a/application/common/ReferenceDispatchService.php +++ b/application/common/ReferenceDispatchService.php @@ -7,7 +7,8 @@ use think\Env; use think\Queue; /** - * 参考文献分流处理:freshRefers 入口之后,按 journal / book / other 三路处理。 + * 参考文献分流处理:按 journal / book / other 三路处理。 + * 每条文献单独入 ReferenceDispatchQueue,由 worker 并行消费。 * * - journal:有 DOI 入 ArticleReferDetailQueue;无 DOI 保留原文 * - book:填充结构化字段 author/title/joura/dateno/isbn(与 Preaccept、References 约定一致) @@ -24,9 +25,9 @@ class ReferenceDispatchService } /** - * 对某篇生产文章的全部参考文献按类型分流处理 + * 将文章下全部有效参考文献逐条入队分流(HTTP 内只做 push,立即返回) */ - public function dispatchRefersByType($pArticleId) + public function enqueueRefersByType($pArticleId) { $pArticleId = intval($pArticleId); if ($pArticleId <= 0) { @@ -43,43 +44,90 @@ class ReferenceDispatchService return; } + foreach ($refers as $refer) { + Queue::push( + 'app\api\job\ReferenceDispatchQueue@fire', + $refer, + 'ReferenceDispatchQueue' + ); + } + } + + /** + * 处理单条参考文献分流(由 ReferenceDispatchQueue 调用) + */ + public function dispatchReferByType($pReferId) + { + $pReferId = intval($pReferId); + if ($pReferId <= 0) { + return; + } + + $refer = Db::name('production_article_refer') + ->where('p_refer_id', $pReferId) + ->where('state', 0) + ->find(); + + if (empty($refer)) { + return; + } + $crossref = new CrossrefService([ 'mailto' => trim((string)Env::get('crossref_mailto', '')), ]); - foreach ($refers as $refer) { - $summary = null; - $crossrefType = ''; - if (trim((string)$refer['refer_doi']) !== '') { - $doiNorm = $this->normalizeDoi($refer['refer_doi']); - if ($doiNorm !== '') { - $summary = $crossref->fetchWorkSummary($doiNorm); - if ($summary && !empty($summary['raw']['type'])) { - $crossrefType = (string)$summary['raw']['type']; - } + $summary = null; + $crossrefType = ''; + if (trim((string)$refer['refer_doi']) !== '') { + $doiNorm = $this->normalizeDoi($refer['refer_doi']); + if ($doiNorm !== '') { + $summary = $crossref->fetchWorkSummary($doiNorm); + if ($summary && !empty($summary['raw']['type'])) { + $crossrefType = (string)$summary['raw']['type']; } } + } - $typeInfo = $this->classifier->classify((string)$refer['refer_content'], $crossrefType); - $dispatchType = $this->classifier->normalizeDispatchType($typeInfo['type']); + $typeInfo = $this->classifier->classify((string)$refer['refer_content'], $crossrefType); + $dispatchType = $this->classifier->normalizeDispatchType($typeInfo['type']); - Db::name('production_article_refer')->where('p_refer_id', $refer['p_refer_id'])->update([ - 'refer_type' => $dispatchType, - 'update_time' => time(), - ]); - $refer['refer_type'] = $dispatchType; + Db::name('production_article_refer')->where('p_refer_id', $refer['p_refer_id'])->update([ + 'refer_type' => $dispatchType, + 'update_time' => time(), + ]); + $refer['refer_type'] = $dispatchType; - switch ($dispatchType) { - case ReferenceTypeClassifier::TYPE_BOOK: - $this->processBookRefer($refer, $summary, $crossref); - break; - case ReferenceTypeClassifier::TYPE_OTHER: - $this->processOtherRefer($refer); - break; - default: - $this->processJournalRefer($refer); - break; - } + switch ($dispatchType) { + case ReferenceTypeClassifier::TYPE_BOOK: + $this->processBookRefer($refer, $summary, $crossref); + break; + case ReferenceTypeClassifier::TYPE_OTHER: + $this->processOtherRefer($refer); + break; + default: + $this->processJournalRefer($refer); + break; + } + } + + /** + * 同步处理整篇文章(仅供脚本/调试,勿在 HTTP 或单条队列任务中批量调用) + */ + public function dispatchRefersByType($pArticleId) + { + $pArticleId = intval($pArticleId); + if ($pArticleId <= 0) { + return; + } + + $refers = Db::name('production_article_refer') + ->where('p_article_id', $pArticleId) + ->where('state', 0) + ->order('index asc, p_refer_id asc') + ->column('p_refer_id'); + + foreach ($refers as $pReferId) { + $this->dispatchReferByType($pReferId); } }